LES CDN prototype

cmd/geth/config.go

@@ -178,6 +178,9 @@ func makeFullNode(ctx *cli.Context) *node.Node {
 	if cfg.Ethstats.URL != "" {
 		utils.RegisterEthStatsService(stack, cfg.Ethstats.URL)
 	}
+	// Add the light server CDN if requested
+	utils.RegisterLesCDNService(stack)
+
 	return stack
 }
 

cmd/utils/flags.go

@@ -49,6 +49,7 @@ import (
 	"github.com/ethereum/go-ethereum/ethstats"
 	"github.com/ethereum/go-ethereum/graphql"
 	"github.com/ethereum/go-ethereum/les"
+	"github.com/ethereum/go-ethereum/lescdn"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/metrics/influxdb"
@@ -1621,6 +1622,19 @@ func RegisterGraphQLService(stack *node.Node, endpoint string, cors, vhosts []st
 	}
 }
 
+// RegisterLesCDNService configures the light client CDN and adds it to the given node.
+func RegisterLesCDNService(stack *node.Node) {
+	if err := stack.Register(func(ctx *node.ServiceContext) (node.Service, error) {
+		var backend *eth.Ethereum
+		if err := ctx.Service(&backend); err == nil {
+			return lescdn.New(backend.BlockChain()), nil
+		}
+		return nil, errors.New("no Ethereum service")
+	}); err != nil {
+		Fatalf("Failed to register the light client CDN service: %v", err)
+	}
+}
+
 func SetupMetrics(ctx *cli.Context) {
 	if metrics.Enabled {
 		log.Info("Enabling metrics collection")

lescdn/chain.go

@@ -0,0 +1,72 @@
+package lescdn
+
+import (
+	"fmt"
+	"net/http"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/hexutil"
+)
+
+// serveChain is responsible for serving HTTP requests for chain data.
+func (s *Service) serveChain(w http.ResponseWriter, r *http.Request) {
+	hash, err := hexutil.Decode(shift(&r.URL.Path))
+	if err != nil {
+		http.Error(w, fmt.Sprintf("invalid block hash: %v", err), http.StatusBadRequest)
+		return
+	}
+	if len(hash) != common.HashLength {
+		http.Error(w, fmt.Sprintf("invalid block hash: length %d != %d", len(hash), common.HashLength), http.StatusBadRequest)
+		return
+	}
+	s.serveChainItem(common.BytesToHash(hash)).ServeHTTP(w, r)
+}
+
+// serveChainItem is responsible for creating a server for HTTP requests for some
+// component of a chain item.
+func (s *Service) serveChainItem(hash common.Hash) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch shift(&r.URL.Path) {
+		case "header":
+			// Retrieve the header and attempt to return it
+			if header := s.chain.GetHeaderByHash(hash); header != nil {
+				reply(w, header)
+				return
+			}
+			// Header not found, error out appropriately
+			http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
+			return
+
+		case "uncles":
+			// Retrieve the block and attempt to return the uncles
+			if block := s.chain.GetBlockByHash(hash); block != nil {
+				reply(w, block.Uncles())
+				return
+			}
+			// Block not found, error out appropriately
+			http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
+			return
+
+		case "transactions":
+			// Retrieve the block and attempt to return the transactions
+			if block := s.chain.GetBlockByHash(hash); block != nil {
+				reply(w, block.Transactions())
+				return
+			}
+			// Block not found, error out appropriately
+			http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
+			return
+
+		case "receipts":
+			// Retrieve the receipts and attempt to return them
+			if receipts := s.chain.GetReceiptsByHash(hash); receipts != nil {
+				reply(w, receipts)
+				return
+			}
+			// Receipts not found, error out appropriately
+			http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
+			return
+		}
+		http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
+	})
+}

lescdn/gzip.go

@@ -0,0 +1,50 @@
+package lescdn
+
+import (
+	"compress/gzip"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"strings"
+	"sync"
+)
+
+var gzPool = sync.Pool{
+	New: func() interface{} {
+		w := gzip.NewWriter(ioutil.Discard)
+		return w
+	},
+}
+
+type gzipResponseWriter struct {
+	io.Writer
+	http.ResponseWriter
+}
+
+func (w *gzipResponseWriter) WriteHeader(status int) {
+	w.Header().Del("Content-Length")
+	w.ResponseWriter.WriteHeader(status)
+}
+
+func (w *gzipResponseWriter) Write(b []byte) (int, error) {
+	return w.Writer.Write(b)
+}
+
+func newGzipHandler(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") {
+			next.ServeHTTP(w, r)
+			return
+		}
+
+		w.Header().Set("Content-Encoding", "gzip")
+
+		gz := gzPool.Get().(*gzip.Writer)
+		defer gzPool.Put(gz)
+
+		gz.Reset(w)
+		defer gz.Close()
+
+		next.ServeHTTP(&gzipResponseWriter{ResponseWriter: w, Writer: gz}, r)
+	})
+}

lescdn/lesleech/main.go

@@ -0,0 +1,147 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"math/big"
+	"net/http"
+	"strconv"
+
+	"github.com/bsipos/thist"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/prque"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/ethereum/go-ethereum/rlp"
+	"github.com/ethereum/go-ethereum/trie"
+)
+
+var (
+	target  = flag.Int("target", 16, "Target number of nodes to pack into a tile")
+	limit   = flag.Int("limit", 256, "Maximum number of nodes to pack into a tile")
+	barrier = flag.Int("barrier", 2, "Trie depth barrier to start tiles at")
+)
+
+func main() {
+	flag.Parse()
+
+	rpc, err := ethclient.Dial("http://127.0.0.1:8545")
+	if err != nil {
+		panic(err)
+	}
+	number, err := strconv.Atoi(flag.Args()[0])
+	if err != nil {
+		panic(err)
+	}
+	previous := make(map[common.Hash]common.StorageSize)
+	for ; ; number++ {
+		// Crawl the entire state trie
+		header, err := rpc.HeaderByNumber(context.Background(), big.NewInt(int64(number)))
+		if err != nil {
+			panic(err)
+		}
+		queue := prque.New(nil)
+		queue.Push(header.Root, 0)
+
+		var (
+			tiles   int
+			storage common.StorageSize
+		)
+		depths := make(map[common.Hash]int)
+
+		tileHists := make(map[int]*thist.Hist)
+		tileCounts := make(map[int]int)
+		tileStorage := make(map[int]common.StorageSize)
+
+		current := make(map[common.Hash]common.StorageSize)
+		for !queue.Empty() {
+			hash, prio := queue.Pop()
+			root, depth := hash.(common.Hash), -prio
+
+			// Read the next tile and dump some statistics
+			nodes, size := fetchTile(root)
+			current[root] = common.StorageSize(size)
+
+			storage += common.StorageSize(size)
+			tiles++
+
+			tileStorage[int(depth)] += common.StorageSize(size)
+			tileCounts[int(depth)]++
+			if _, ok := tileHists[int(depth)]; !ok {
+				tileHists[int(depth)] = thist.NewHist(nil, "", "auto", -1, true)
+			}
+			tileHists[int(depth)].Update(float64(size))
+
+			log.Printf("D=%d R=0x%x: %d nodes, %v (%d tiles, %v total)", depth, root, len(nodes), common.StorageSize(size), tiles, storage)
+
+			// Decode the tile and continue expansion
+			pulled := make(map[common.Hash]struct{})
+			for _, node := range nodes {
+				pulled[crypto.Keccak256Hash(node)] = struct{}{}
+			}
+			for _, node := range nodes {
+				trie.IterateRefs(node, func(path []byte, child common.Hash) error {
+					depths[child] = depths[crypto.Keccak256Hash(node)] + len(path)
+					if _, ok := pulled[child]; !ok {
+						queue.Push(child, -int64(depths[child]))
+					}
+					return nil
+				})
+			}
+		}
+		// Report any tile stats
+		fmt.Println("\nTile stats:")
+		for i := 0; i < 128; i++ {
+			if tileCounts[i] > 0 {
+				fmt.Printf("  Depth %d: %d tiles, %v\n", i, tileCounts[i], tileStorage[i])
+				fmt.Println(tileHists[i].Draw())
+			}
+		}
+		// Compare the current tileset with the previous one and report the diff
+		var (
+			addSize, dupSize, delSize    common.StorageSize
+			addCount, dupCount, delCount int
+		)
+		for hash, size := range current {
+			if _, ok := previous[hash]; ok {
+				dupSize += size
+				dupCount++
+			} else {
+				addSize += size
+				addCount++
+			}
+		}
+		for hash, size := range previous {
+			if _, ok := current[hash]; !ok {
+				delSize += size
+				delCount++
+			}
+		}
+		previous = current
+
+		fmt.Printf("Block %d: Added %d(%v), removed %d(%v), retained %d(%v)\n", number, addCount, addSize, delCount, delSize, dupCount, dupSize)
+	}
+}
+
+// fetchTile retrieves a tile rooted at a certain trie hash node, also returning
+// the number of bytes the transfered raw data consisted of.
+func fetchTile(root common.Hash) ([][]byte, int) {
+	res, err := http.Get(fmt.Sprintf("http://127.0.0.1:8548/state/0x%x?target=%d&limit=%d&barrier=%d", root, *target, *limit, *barrier))
+	if err != nil {
+		panic(err)
+	}
+	blob, err := ioutil.ReadAll(res.Body)
+	if err != nil {
+		panic(err)
+	}
+	res.Body.Close()
+
+	var nodes [][]byte
+	if err := rlp.DecodeBytes(blob, &nodes); err != nil {
+		panic(err)
+	}
+	return nodes, len(blob)
+}

lescdn/service.go

@@ -0,0 +1,89 @@
+package lescdn
+
+import (
+	"net/http"
+	"path"
+	"strings"
+
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/rlp"
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+// Service is a RESTful HTTP service meant to act as a data source for a light client
+// content distribution network.
+type Service struct {
+	chain *core.BlockChain
+}
+
+// New creates a data source for a les content distribution network.
+func New(chain *core.BlockChain) *Service {
+	return &Service{
+		chain: chain,
+	}
+}
+
+// Protocols implements node.Service, returning the P2P network protocols used
+// by the lescdn service (nil as it doesn't use the devp2p overlay network).
+func (s *Service) Protocols() []p2p.Protocol { return nil }
+
+// APIs implements node.Service, returning the RPC API endpoints provided by the
+// lescdn service (nil as it doesn't provide any user callable APIs).
+func (s *Service) APIs() []rpc.API { return nil }
+
+// Start implements node.Service, starting up the content distribution source.
+func (s *Service) Start(server *p2p.Server) error {
+	go http.ListenAndServe("localhost:8548", newGzipHandler(s))
+
+	log.Info("Light client CDN started")
+	return nil
+}
+
+// Stop implements node.Service, terminating the content distribution source.
+func (s *Service) Stop() error {
+	log.Info("Light client CDN stopped")
+	return nil
+}
+
+// ServeHTTP is the entry point of the les cdn, splitting the request across the
+// supported submodules.
+func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	switch shift(&r.URL.Path) {
+	case "chain":
+		s.serveChain(w, r)
+		return
+
+	case "state":
+		s.serveState(w, r)
+		return
+	}
+	http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
+}
+
+// shift splits off the first component of p, which will be cleaned of relative
+// components before processing. The returned head will never contain a slash and
+// the remaining tail will always be a rooted path without trailing slash.
+func shift(p *string) string {
+	*p = path.Clean("/" + *p)
+
+	var head string
+	if idx := strings.Index((*p)[1:], "/") + 1; idx > 0 {
+		head = (*p)[1:idx]
+		*p = (*p)[idx:]
+	} else {
+		head = (*p)[1:]
+		*p = "/"
+	}
+	return head
+}
+
+// reply marshals a value into the response stream via RLP, also setting caching
+// to indefinite.
+func reply(w http.ResponseWriter, v interface{}) {
+	w.Header().Set("Cache-Control", "max-age=31536000") // 1 year cache expiry
+	if err := rlp.Encode(w, v); err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+	}
+}

lescdn/state.go

@@ -0,0 +1,152 @@
+package lescdn
+
+import (
+	"fmt"
+	"net/http"
+	"strconv"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/common/prque"
+	"github.com/ethereum/go-ethereum/trie"
+)
+
+const (
+	// tileTarget is the target number of trie nodes to place into each tile. The
+	// actual count might be smaller for leaf tiles, or larger to ensure proper
+	// tile barriers.
+	tileTarget = 16
+
+	// tileLimit is the maximum number of trie nodes to place into each tile, above
+	// which the tile is forcefully split, even if that means breaking barriers.
+	tileLimit = 256
+
+	// tileBarrier is the trie depth multiplier where trie nodes need to end to
+	// ensure that mutating tries still reuse the same non-mutated tiles.
+	//
+	// The number 2 was chosen experimentally, but the rationalization behind it is
+	// that nodes close to the root will be very dense. The worst case where the nodes
+	// are filled, a barrier of 3 would result in about 16^3 = 4096 hash pointers in
+	// the leaves, which is 128KB + internal pointers + nodes + boilerplate. Depending
+	// on trie shape, this can grow to even larger values, becoming useless, especially
+	// at the root, so 2 seems to be a limit. A barrier of 2 produced about 8KB tiles
+	// in our experiments.
+	tileBarrier = 2
+)
+
+// serveState is responsible for serving HTTP requests for state data.
+func (s *Service) serveState(w http.ResponseWriter, r *http.Request) {
+	// TODO(karalabe): the non-defaults are for benchmarking, get rid when finalized
+	cutTileTarget := int64(tileTarget)
+	if target, ok := r.URL.Query()["target"]; ok {
+		cutTileTarget, _ = strconv.ParseInt(target[0], 0, 64)
+	}
+	curTileLimit := int64(tileLimit)
+	if limit, ok := r.URL.Query()["limit"]; ok {
+		curTileLimit, _ = strconv.ParseInt(limit[0], 0, 64)
+	}
+	curTileBarrier := int64(tileBarrier)
+	if barrier, ok := r.URL.Query()["barrier"]; ok {
+		curTileBarrier, _ = strconv.ParseInt(barrier[0], 0, 64)
+	}
+	// Decode the root of the subtrie tile we should return
+	root, err := hexutil.Decode(shift(&r.URL.Path))
+	if err != nil {
+		http.Error(w, fmt.Sprintf("invalid state root: %v", err), http.StatusBadRequest)
+		return
+	}
+	if len(root) != common.HashLength {
+		http.Error(w, fmt.Sprintf("invalid state root: length %d != %d", len(root), common.HashLength), http.StatusBadRequest)
+		return
+	}
+	// Do a breadth-first expansion to collect a fixed size tile
+	triedb := s.chain.StateCache().TrieDB()
+
+	nodes, refset, cutset, err := makeIdealTile(triedb, common.BytesToHash(root), int(cutTileTarget), int(curTileBarrier))
+	if err != nil {
+		http.Error(w, fmt.Sprintf("failed to make tile: %v", err), http.StatusBadRequest)
+		return
+	}
+	// If our cutset nodes won't result in meaningful tiles (they reach the leaves),
+	// merge all of them into the current tile to avoid creating millions of subtiles.
+	var (
+		merged []common.Hash
+		merges [][]byte
+	)
+	for !cutset.Empty() {
+		// Fetch the deepest cutset node and merge in if it's a leaf
+		hash := cutset.PopItem().(common.Hash)
+
+		subnodes, _, subcutset, err := makeIdealTile(triedb, hash, int(cutTileTarget), int(curTileBarrier))
+		if err != nil {
+			http.Error(w, fmt.Sprintf("failed to make subtile: %v", err), http.StatusBadRequest)
+			return
+		}
+		if subcutset.Empty() {
+			merged = append(merged, hash)
+			merges = append(merges, subnodes...)
+			continue
+		}
+		// Deepest cutset node produces non-leaf tile, don't bother with shallower node
+		break
+	}
+	// If the final tile became huge, it means we packed in too many leaves due to
+	// tile mergers. Shave off the nodes that caused tile mergers in the first place.
+	if len(nodes)+len(merges) > int(curTileLimit) {
+		for _, drop := range merged {
+			for i, refs := range refset {
+				if _, ok := refs[drop]; ok {
+					nodes = append(nodes[:i], nodes[i+1:]...)
+					refset = append(refset[:i], refset[i+1:]...)
+					break
+				}
+			}
+		}
+	} else {
+		nodes = append(nodes, merges...)
+	}
+	reply(w, nodes)
+}
+
+// makeIdealTile gathers trie nodes and assembles an ideal tile: one that barely
+// exceeds the allowed node count and terminates at tile boundaries.
+func makeIdealTile(triedb *trie.Database, root common.Hash, limit int, barrier int) ([][]byte, []map[common.Hash]struct{}, *prque.Prque, error) {
+	queue := prque.New(nil)
+	queue.Push(root, 0)
+
+	var (
+		nodes  [][]byte
+		refset []map[common.Hash]struct{}
+		cutset = prque.New(nil)
+	)
+	for !queue.Empty() {
+		// Fetch the next trie node, which may or may not be included in the tile
+		root, prio := queue.Pop()
+		hash, depth := root.(common.Hash), -prio
+
+		if len(nodes) > int(limit) {
+			// Tile exceeded its recommended size. If the next node is on a tile barrier,
+			// leave it to be collected in a next run (or retrieved from a cache).
+			if int(depth)%barrier == 0 {
+				cutset.Push(hash, depth)
+				continue
+			}
+		}
+		// Tile not done yet, fetch the next node and append it to the tile
+		node, err := triedb.Node(hash)
+		if err != nil {
+			return nil, nil, nil, err
+		}
+		nodes = append(nodes, node)
+
+		// Expand the trie node and queue all children up
+		refs := make(map[common.Hash]struct{})
+		trie.IterateRefs(node, func(path []byte, child common.Hash) error {
+			queue.Push(child, -(depth + int64(len(path))))
+			refs[child] = struct{}{}
+			return nil
+		})
+		refset = append(refset, refs)
+	}
+	return nodes, refset, cutset, nil
+}

trie/iterator.go

@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"container/heap"
 	"errors"
+	"fmt"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/rlp"
@@ -575,3 +576,35 @@ func (it *unionIterator) Error() error {
 	}
 	return nil
 }
+
+// IterateRefs decodes a trie node, iterates all its children and invokes a
+// callback for each hash node found.
+func IterateRefs(node []byte, onHashNode func([]byte, common.Hash) error) error {
+	return iterateRefs(mustDecodeNode(nil, node), nil, onHashNode)
+}
+
+// iterateRefs traverses the node hierarchy of a cached node and invokes the
+// provided callback on all hash nodes.
+func iterateRefs(n node, path []byte, onHashNode func([]byte, common.Hash) error) error {
+	switch n := n.(type) {
+	case *shortNode:
+		return iterateRefs(n.Val, append(path, n.Key...), onHashNode)
+
+	case *fullNode:
+		for i := 0; i < 16; i++ {
+			if err := iterateRefs(n.Children[i], append(path, byte(i)), onHashNode); err != nil {
+				return err
+			}
+		}
+		return nil
+
+	case hashNode:
+		return onHashNode(path, common.BytesToHash(n))
+
+	case valueNode, nil:
+		return nil
+
+	default:
+		panic(fmt.Sprintf("unknown node type: %T", n))
+	}
+}