diff --git a/cmd/geth/config.go b/cmd/geth/config.go index 71e8355348..07006b575f 100644 --- a/cmd/geth/config.go +++ b/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 } diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index bdadebd852..5ede9de434 100644 --- a/cmd/utils/flags.go +++ b/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") diff --git a/lescdn/chain.go b/lescdn/chain.go new file mode 100644 index 0000000000..1a71dc1362 --- /dev/null +++ b/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) + }) +} diff --git a/lescdn/gzip.go b/lescdn/gzip.go new file mode 100644 index 0000000000..a21d4beb73 --- /dev/null +++ b/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) + }) +} diff --git a/lescdn/lesleech/main.go b/lescdn/lesleech/main.go new file mode 100644 index 0000000000..d5133d6abf --- /dev/null +++ b/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) +} diff --git a/lescdn/service.go b/lescdn/service.go new file mode 100644 index 0000000000..16994ad7ed --- /dev/null +++ b/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) + } +} diff --git a/lescdn/state.go b/lescdn/state.go new file mode 100644 index 0000000000..b4f8761d04 --- /dev/null +++ b/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 +} diff --git a/trie/iterator.go b/trie/iterator.go index bb4025d8f3..40b7ba6190 100644 --- a/trie/iterator.go +++ b/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)) + } +}