Blocks compression (#510)

* block compression

* rerun lint

* fix lint

common/debug/experiments.go

@@ -2,9 +2,15 @@ package debug
 
 import (
 	"os"
+	"sync"
 	"sync/atomic"
 )
 
+var (
+	compressBlocks    bool
+	getCompressBlocks sync.Once
+)
+
 // atomic: bit 0 is the value, bit 1 is the initialized flag
 var getNodeData uint32
 
@@ -40,3 +46,10 @@ func OverrideGetNodeData(val bool) {
 		atomic.StoreUint32(&getNodeData, gndInitializedFlag)
 	}
 }
+
+func IsBlockCompressionEnabled() bool {
+	getCompressBlocks.Do(func() {
+		_, compressBlocks = os.LookupEnv("COMPRESS_BLOCKS")
+	})
+	return compressBlocks
+}

core/rawdb/accessors_chain.go

@@ -20,6 +20,8 @@ import (
 	"bytes"
 	"context"
 	"encoding/binary"
+	"github.com/golang/snappy"
+	"github.com/ledgerwatch/turbo-geth/common/debug"
 	"math/big"
 
 	"github.com/ledgerwatch/turbo-geth/common"
@@ -239,6 +241,13 @@ func deleteHeaderWithoutNumber(db DatabaseDeleter, hash common.Hash, number uint
 // ReadBodyRLP retrieves the block body (transactions and uncles) in RLP encoding.
 func ReadBodyRLP(db DatabaseReader, hash common.Hash, number uint64) rlp.RawValue {
 	data, _ := db.Get(dbutils.BlockBodyPrefix, dbutils.BlockBodyKey(number, hash))
+	if debug.IsBlockCompressionEnabled() && len(data) > 0 {
+		var err error
+		data, err = snappy.Decode(nil, data)
+		if err != nil {
+			log.Warn("err on decode block", "err", err)
+		}
+	}
 	return data
 }
 
@@ -247,6 +256,9 @@ func WriteBodyRLP(ctx context.Context, db DatabaseWriter, hash common.Hash, numb
 	if common.IsCanceled(ctx) {
 		return
 	}
+	if debug.IsBlockCompressionEnabled() {
+		rlp = snappy.Encode(nil, rlp)
+	}
 	if err := db.Put(dbutils.BlockBodyPrefix, dbutils.BlockBodyKey(number, hash), rlp); err != nil {
 		log.Crit("Failed to store block body", "err", err)
 	}

core/rawdb/accessors_chain_test.go

@@ -21,15 +21,14 @@ import (
 	"context"
 	"encoding/hex"
 	"fmt"
-	"math/big"
-	"testing"
-
 	"github.com/ledgerwatch/turbo-geth/common"
 	"github.com/ledgerwatch/turbo-geth/core/types"
 	"github.com/ledgerwatch/turbo-geth/ethdb"
 	"github.com/ledgerwatch/turbo-geth/params"
 	"github.com/ledgerwatch/turbo-geth/rlp"
 	"golang.org/x/crypto/sha3"
+	"math/big"
+	"testing"
 )
 
 // Tests block header storage and retrieval operations.

eth/handler_test.go

@@ -23,6 +23,7 @@ import (
 	"math"
 	"math/big"
 	"math/rand"
+	"strconv"
 	"testing"
 	"time"
 
@@ -178,29 +179,39 @@ func testGetBlockHeaders(t *testing.T, protocol int) {
 	}
 	// Run each of the tests and verify the results against the chain
 	for i, tt := range tests {
-		peer, _ := newTestPeer("peer", protocol, pm, true)
-		defer peer.close()
-		// Collect the headers to expect in the response
-		headers := []*types.Header{}
-		for _, hash := range tt.expect {
-			headers = append(headers, pm.blockchain.GetBlockByHash(hash).Header())
-		}
-		// Send the hash request and verify the response
-		p2p.Send(peer.app, 0x03, tt.query)
-		if err := p2p.ExpectMsg(peer.app, 0x04, headers); err != nil {
-			t.Errorf("test %d: headers mismatch: %v", i, err)
-		}
-		// If the test used number origins, repeat with hashes as the too
-		if tt.query.Origin.Hash == (common.Hash{}) {
-			if origin := pm.blockchain.GetBlockByNumber(tt.query.Origin.Number); origin != nil {
-				tt.query.Origin.Hash, tt.query.Origin.Number = origin.Hash(), 0
+		i := i
+		tt := tt
+		t.Run(strconv.Itoa(i), func(t *testing.T) {
+			peer, _ := newTestPeer("peer", protocol, pm, true)
+			defer peer.close()
+			// Collect the headers to expect in the response
+			headers := []*types.Header{}
+			for _, hash := range tt.expect {
+				headers = append(headers, pm.blockchain.GetBlockByHash(hash).Header())
+			}
+			// Send the hash request and verify the response
+			if err := p2p.Send(peer.app, 0x03, tt.query); err != nil {
+				t.Error(err)
+			}
+			if err := p2p.ExpectMsg(peer.app, 0x04, headers); err != nil {
+				t.Errorf("test %d: headers mismatch: %v", i, err)
+			}
+			// If the test used number origins, repeat with hashes as the too
+			if tt.query.Origin.Hash == (common.Hash{}) {
+				if origin := pm.blockchain.GetBlockByNumber(tt.query.Origin.Number); origin != nil {
+					tt.query.Origin.Hash, tt.query.Origin.Number = origin.Hash(), 0
 
-				p2p.Send(peer.app, 0x03, tt.query)
-				if err := p2p.ExpectMsg(peer.app, 0x04, headers); err != nil {
-					t.Errorf("test %d: headers mismatch: %v", i, err)
+					if err := p2p.Send(peer.app, 0x03, tt.query); err != nil {
+						t.Error(err)
+					}
+
+					if err := p2p.ExpectMsg(peer.app, 0x04, headers); err != nil {
+						t.Errorf("test %d: headers mismatch: %v", i, err)
+					}
 				}
 			}
-		}
+
+		})
 	}
 }
 

ethdb/remote/remotechain/chain_remote.go

@@ -4,6 +4,8 @@ import (
 	"bytes"
 	"encoding/binary"
 	"fmt"
+	"github.com/golang/snappy"
+	"github.com/ledgerwatch/turbo-geth/common/debug"
 	"math/big"
 
 	"github.com/ledgerwatch/turbo-geth/common"
@@ -115,6 +117,15 @@ func ReadBodyRLP(tx ethdb.Tx, hash common.Hash, number uint64) (rlp.RawValue, er
 	if bucket == nil {
 		return rlp.RawValue{}, fmt.Errorf("bucket %s not found", dbutils.HeaderPrefix)
 	}
+
+	if debug.IsBlockCompressionEnabled() {
+		data, err := bucket.Get(dbutils.BlockBodyKey(number, hash))
+		if err != nil {
+			return nil, err
+		}
+		return snappy.Decode(nil, data)
+	}
+
 	return bucket.Get(dbutils.BlockBodyKey(number, hash))
 }