From 7eeb79ebba9e979e5594a354a4d0e634acba0a6e Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Wed, 7 Nov 2018 09:17:05 -0500 Subject: [PATCH] swarm: swap accounting for swarm with p2p accounting --- swarm/network/stream/stream.go | 40 +-- swarm/swap_test.go | 429 --------------------------------- swarm/swarm.go | 5 +- 3 files changed, 2 insertions(+), 472 deletions(-) delete mode 100644 swarm/swap_test.go diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 26b74b155c..03a6b1edb4 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -88,9 +88,9 @@ type Registry struct { intervalsStore state.Store autoRetrieval bool //automatically subscribe to retrieve request stream maxPeerServers int + spec *protocols.Spec //this protocol's spec balance protocols.Balance //implements protocols.Balance, for accounting prices protocols.Prices //implements protocols.Prices, provides prices to accounting - spec *protocols.Spec //this protocol's spec } // RegistryOptions holds optional values for NewRegistry constructor. @@ -235,14 +235,10 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy return streamer } -//This is an accounted protocol, therefore we need to provide a pricing Hook to the spec -//For simulations to be able to run multiple nodes and not override the hook's balance, //we need to construct a spec instance per node instance func (r *Registry) setupSpec() { //first create the "bare" spec r.createSpec() - //now create the pricing object - r.createPriceOracle() //if balance is nil, this node has been started without swap support (swapEnabled flag is false) if r.balance != nil && !reflect.ValueOf(r.balance).IsNil() { //swap is enabled, so setup the hook @@ -766,40 +762,6 @@ func (r *Registry) createSpec() { r.spec = spec } -//An accountable message needs some meta information attached to it -//in order to evaluate the correct price -type StreamerPrices struct { - priceMatrix map[reflect.Type]*protocols.Price - registry *Registry -} - -func (spo *StreamerPrices) Price(msg interface{}) *protocols.Price { - typ := reflect.TypeOf(msg).Elem() - return spo.priceMatrix[typ] -} - -func (r *Registry) createPriceOracle() { - - po := &StreamerPrices{ - registry: r, - } - po.priceMatrix = map[reflect.Type]*protocols.Price{ - - reflect.TypeOf(ChunkDeliveryMsgRetrieval{}): &protocols.Price{ - Value: uint64(100), - PerByte: true, - Payer: protocols.Receiver, - }, - - reflect.TypeOf(RetrieveRequestMsg{}): &protocols.Price{ - Value: uint64(10), - PerByte: false, - Payer: protocols.Sender, - }, - } - r.prices = po -} - func (r *Registry) Protocols() []p2p.Protocol { return []p2p.Protocol{ { diff --git a/swarm/swap_test.go b/swarm/swap_test.go deleted file mode 100644 index 8a3faee10b..0000000000 --- a/swarm/swap_test.go +++ /dev/null @@ -1,429 +0,0 @@ -// Copyright 2018 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package swarm - -import ( - "context" - "errors" - "fmt" - "io/ioutil" - "math/rand" - "os" - "strconv" - "sync" - "testing" - "time" - - "github.com/ethereum/go-ethereum/crypto" - "github.com/ethereum/go-ethereum/metrics" - "github.com/ethereum/go-ethereum/node" - "github.com/ethereum/go-ethereum/p2p/enode" - "github.com/ethereum/go-ethereum/p2p/simulations/adapters" - "github.com/ethereum/go-ethereum/swarm/api" - "github.com/ethereum/go-ethereum/swarm/log" - "github.com/ethereum/go-ethereum/swarm/network/simulation" - "github.com/ethereum/go-ethereum/swarm/storage" -) - -//In TestSwapNetworkSymmetricFileUpload we set up a network with arbitrary number of nodes -//(16), and each of the nodes uploads a file of same size -//Afterwards we check that every node's balance WITH ANOTHER PEER -//has the same value but opposite sign -func TestSwapNetworkSymmetricFileUpload(t *testing.T) { - //default hardcoded network size - nodeCount := 16 - - //setup the simulation - //use a complete node setup via `NewSwam` - sim := simulation.New(map[string]simulation.ServiceFunc{ - "swarm": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) { - config := api.NewConfig() - config.Port = strconv.Itoa(8500 + rand.Intn(9999)) - - dir, err := ioutil.TempDir("", "swap-network-test-node"+config.Port) - if err != nil { - return nil, nil, err - } - cleanup = func() { - err := os.RemoveAll(dir) - if err != nil { - log.Error("cleaning up swarm temp dir", "err", err) - } - } - - config.Path = dir - - privkey, err := crypto.GenerateKey() - if err != nil { - return nil, cleanup, err - } - - config.Init(privkey) - - //set Swap to be enabled for this test - config.SwapEnabled = true - - swarm, err := NewSwarm(config, nil) - if err != nil { - return nil, cleanup, err - } - - bucket.Store(bucketKeySwarm, swarm) - log.Info("new swarm", "bzzKey", config.BzzKey, "baseAddr", fmt.Sprintf("%x", swarm.bzz.BaseAddr())) - return swarm, cleanup, nil - }, - }) - defer sim.Close() - - ctx := context.Background() - files := make([]file, 0) - - var checkStatusM sync.Map - var nodeStatusM sync.Map - var totalFoundCount uint64 - - //connect all nodes in a chain - _, err := sim.AddNodesAndConnectChain(nodeCount) - if err != nil { - t.Fatal(err) - } - - //run the simulation - result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error { - //wait for kademlia to be healthy - if _, err := sim.WaitTillHealthy(ctx, 2); err != nil { - return err - } - - nodeIDs := sim.UpNodeIDs() - shuffle(len(nodeIDs), func(i, j int) { - nodeIDs[i], nodeIDs[j] = nodeIDs[j], nodeIDs[i] - }) - //upload a file for every node - for _, id := range nodeIDs { - item, ok := sim.NodeItem(id, bucketKeySwarm) - if !ok { - log.Error("No swarm") - return errors.New("No swarm") - } - swarm := item.(*Swarm) - key, data, err := uploadFile(swarm) - if err != nil { - return err - } - log.Trace("file uploaded", "node", id, "key", key.String()) - files = append(files, file{ - addr: key, - data: data, - nodeID: id, - }) - } - - // File retrieval check is repeated until all uploaded files are retrieved from all nodes - // or until the timeout is reached. - for { - if retrieve(sim, files, &checkStatusM, &nodeStatusM, &totalFoundCount) == 0 { - break - } - } - - time.Sleep(5 * time.Second) - //every node has a map to all nodes it had interactions - //each entry in the map is a map of the other node with all the balances - balancesMap := make(map[enode.ID]map[enode.ID]int64) - - //iterate all nodes - for _, node := range sim.NodeIDs() { - item, ok := sim.NodeItem(node, bucketKeySwarm) - if !ok { - log.Error("No swarm") - return errors.New("No swarm") - } - swarm := item.(*Swarm) - - //submap for each node is a map of all nodes with the balance for that node - subBalances := make(map[enode.ID]int64) - - //iterate all nodes again... - //get all balances with other peers for every node - for _, n := range sim.NodeIDs() { - if node == n { - continue - } - - //get the peer's balance with this node - balance, err := swarm.swap.GetPeerBalance(n) - if err == nil { - subBalances[n] = balance - log.Debug(fmt.Sprintf("Balance of node %s to node %s: %d", node.TerminalString(), n.TerminalString(), balance)) - } else { - log.Debug(fmt.Sprintf("Node %s has no balance with node %s", node.TerminalString(), n.TerminalString())) - } - } - //update the map for this node - balancesMap[node] = subBalances - } - - //print all the balances if requested - if *printStats { - for k, v := range balancesMap { - fmt.Println(fmt.Sprintf("node %s balances:", k.TerminalString())) - for kk, vv := range v { - fmt.Println(fmt.Sprintf(".........with node %s: balance %d", kk.TerminalString(), vv)) - } - } - //NOTE: this are currently metrics over ALL nodes, not per node - fmt.Println(fmt.Sprintf("Total units credited: %d", metrics.Get("account.balance.credit").(metrics.Counter).Count())) - fmt.Println(fmt.Sprintf("Total units debited: %d", metrics.Get("account.balance.debit").(metrics.Counter).Count())) - fmt.Println(fmt.Sprintf("Total bytes credited: %d", metrics.Get("account.bytes.credit").(metrics.Counter).Count())) - fmt.Println(fmt.Sprintf("Total bytes debited: %d", metrics.Get("account.bytes.debit").(metrics.Counter).Count())) - //fmt.Println(fmt.Sprintf("Cheques issued: %d", metrics.Get("account.cheques.issued").(metrics.Counter).Count())) - //fmt.Println(fmt.Sprintf("Cheques received: %d", metrics.Get("account.cheques.received").(metrics.Counter).Count())) - fmt.Println(fmt.Sprintf("Number of messages credited: %d", metrics.Get("account.msg.credit").(metrics.Counter).Count())) - fmt.Println(fmt.Sprintf("Number of messages debited: %d", metrics.Get("account.msg.debit").(metrics.Counter).Count())) - fmt.Println(fmt.Sprintf("Peers dropped: %d", metrics.Get("account.peerdrops").(metrics.Counter).Count())) - fmt.Println(fmt.Sprintf("Number of times node dropped itself: %d", metrics.Get("account.selfdrops").(metrics.Counter).Count())) - } - - //now iterate the whole map - //and check that every node k has the same - //balance with a peer as that peer with the node, - //but in inverted signs - - //iterate the map - success := true - for k, mapForK := range balancesMap { - //iterate the submap - for n, balanceKwithN := range mapForK { - //iterate the main map again - for subK, mapForSubK := range balancesMap { - //if the node and the peer are the same... - if n == subK { - log.Trace(fmt.Sprintf("balance of %s with %s: %d", k.TerminalString(), n.TerminalString(), balanceKwithN)) - log.Trace(fmt.Sprintf("balance of %s with %s: %d", n.TerminalString(), k.TerminalString(), mapForSubK[k])) - //...check that they have the same balance in Abs terms and that it is not 0 - if balanceKwithN+mapForSubK[k] != 0 && balanceKwithN != 0 { - log.Error(fmt.Sprintf("Expected balances to be a+b = 0 AND balance(a) != 0, but they are not, balance k with n: %d, balance n with k: %d", balanceKwithN, mapForSubK[k])) - success = false - } - } - } - } - } - if success { - return nil - } - return errors.New("some conditions could not be met") - }) - - if result.Error != nil { - t.Fatal(result.Error) - } - log.Debug("test terminated") -} - -//TestSwapNetworkAsymmetricFileUpload is a swap test too, -//but this time the number and size of files are random -func TestSwapNetworkAsymmetricFileUpload(t *testing.T) { - nodeCount := 16 - - sim := simulation.New(map[string]simulation.ServiceFunc{ - "swarm": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) { - config := api.NewConfig() - config.Port = strconv.Itoa(8500 + rand.Intn(9999)) - - dir, err := ioutil.TempDir("", "swap-network-test-node"+config.Port) - if err != nil { - return nil, nil, err - } - cleanup = func() { - err := os.RemoveAll(dir) - if err != nil { - log.Error("cleaning up swarm temp dir", "err", err) - } - } - - config.Path = dir - - privkey, err := crypto.GenerateKey() - if err != nil { - return nil, cleanup, err - } - - config.Init(privkey) - //enable swap - config.SwapEnabled = true - - swarm, err := NewSwarm(config, nil) - if err != nil { - return nil, cleanup, err - } - bucket.Store(bucketKeySwarm, swarm) - log.Info("new swarm", "bzzKey", config.BzzKey, "baseAddr", fmt.Sprintf("%x", swarm.bzz.BaseAddr())) - return swarm, cleanup, nil - }, - }) - defer sim.Close() - - ctx := context.Background() - files := make([]file, 0) - - var checkStatusM sync.Map - var nodeStatusM sync.Map - var totalFoundCount uint64 - - _, err := sim.AddNodesAndConnectChain(nodeCount) - if err != nil { - t.Fatal(err) - } - - //NOTE: maxFileSize is 4 kB, this in order to provide faster tests - //it would be interesting to run these tests with bigger files - //(to see how drop limits are affected etc.) - const maxFileSize = 1024 * 4 //1024 bytes * 4 = 4kB - const minfileSize = 1024 - - //pseudo random algo to define if a node will upload or not - //if a bit is 0, do not upload - pseudoRandomNum := rand.Int63() - pseudoRandomBitMask := strconv.FormatInt(pseudoRandomNum, 2) - - result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error { - if _, err := sim.WaitTillHealthy(ctx, 2); err != nil { - return err - } - - nodeIDs := sim.UpNodeIDs() - shuffle(len(nodeIDs), func(i, j int) { - nodeIDs[i], nodeIDs[j] = nodeIDs[j], nodeIDs[i] - }) - for i, id := range nodeIDs { - //if the position in random num is 0, don't upload - if string(pseudoRandomBitMask[i]) != "0" { - size := rand.Intn(maxFileSize-minfileSize) + minfileSize - key, data, err := uploadRandomFileSize(sim.Service("swarm", id).(*Swarm), size) - if err != nil { - return err - } - log.Trace("file uploaded", "node", id, "key", key.String()) - files = append(files, file{ - addr: key, - data: data, - nodeID: id, - }) - } - } - - // File retrieval check is repeated until all uploaded files are retrieved from all nodes - // or until the timeout is reached. - for { - if retrieve(sim, files, &checkStatusM, &nodeStatusM, &totalFoundCount) == 0 { - break - } - } - - time.Sleep(5 * time.Second) - - balancesMap := make(map[enode.ID]map[enode.ID]int64) - - for _, node := range sim.NodeIDs() { - item, ok := sim.NodeItem(node, bucketKeySwarm) - if !ok { - log.Error("No swarm") - return errors.New("no swarm") - } - swarm := item.(*Swarm) - - subBalances := make(map[enode.ID]int64) - - for _, n := range sim.NodeIDs() { - if node == n { - continue - } - balance, err := swarm.swap.GetPeerBalance(n) - if err == nil { - subBalances[n] = balance - log.Debug(fmt.Sprintf("Balance of node %s to node %s: %d", node.TerminalString(), n.TerminalString(), balance)) - } else { - log.Debug(fmt.Sprintf("Node %s has no balance with node %s", node.TerminalString(), n.TerminalString())) - } - } - balancesMap[node] = subBalances - } - - if *printStats { - for k, v := range balancesMap { - fmt.Println(fmt.Sprintf("node %s balances:", k.TerminalString())) - for kk, vv := range v { - fmt.Println(fmt.Sprintf(".........with node %s: balance %d", kk.TerminalString(), vv)) - } - } - } - - /* - Assuming that in this case, balances should be symmetric too I - */ - - success := true - for k, mapForK := range balancesMap { - for n, balanceKwithN := range mapForK { - for subK, mapForSubK := range balancesMap { - if n == subK { - log.Trace(fmt.Sprintf("balance of %s with %s: %d", k.TerminalString(), n.TerminalString(), balanceKwithN)) - log.Trace(fmt.Sprintf("balance of %s with %s: %d", n.TerminalString(), k.TerminalString(), mapForSubK[k])) - if balanceKwithN+mapForSubK[k] != 0 && balanceKwithN != 0 { - log.Error(fmt.Sprintf("Expected balances to be a+b = 0 AND balance(a) != 0, but they are not, balance k with n: %d, balance n with k: %d", balanceKwithN, mapForSubK[k])) - success = false - } - } - } - } - } - - if success { - return nil - } - - return errors.New("some conditions could not be met") - }) - - if result.Error != nil { - t.Fatal(result.Error) - } - log.Debug("test terminated") -} - -// uploadFile, uploads a short file to the swarm instance -// using the api.Put method. -func uploadRandomFileSize(swarm *Swarm, size int) (storage.Address, string, error) { - b := make([]byte, size) - _, err := rand.Read(b) - if err != nil { - return nil, "", err - } - // uniqueness is very certain. - data := fmt.Sprintf("test content %s %x", time.Now().Round(0), b) - ctx := context.TODO() - k, wait, err := swarm.api.Put(ctx, data, "text/plain", false) - if err != nil { - return nil, "", err - } - if wait != nil { - err = wait(ctx) - } - return k, data, err -} diff --git a/swarm/swarm.go b/swarm/swarm.go index a128b5ac8b..70cb0898ff 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -51,6 +51,7 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" "github.com/ethereum/go-ethereum/swarm/storage/feed" "github.com/ethereum/go-ethereum/swarm/storage/mock" + "github.com/ethereum/go-ethereum/swarm/swap" "github.com/ethereum/go-ethereum/swarm/tracing" ) @@ -198,10 +199,6 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e SyncUpdateDelay: config.SyncUpdateDelay, MaxPeerServers: config.MaxStreamPeerServers, } - if config.LightNodeEnabled { - registryOptions.DoSync = false - registryOptions.DoRetrieve = false - } self.streamer = stream.NewRegistry(nodeID, delivery, self.netStore, stateStore, registryOptions, self.swap) // Swarm Hash Merklised Chunking for Arbitrary-length Document/File storage