From 55f45d25236f357af80c92839caa96e8ecaec58d Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Thu, 30 Aug 2018 11:49:08 -0500 Subject: [PATCH] swarm/swap: new swap accounting with SwapEnabled flag --- swarm/swap/swap.go | 23 ++++++++++------- swarm/swap_test.go | 64 +++++++++++++++++++++++++++++++++++++--------- swarm/swarm.go | 17 +++++------- 3 files changed, 73 insertions(+), 31 deletions(-) diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go index c31c528906..5b7c29906b 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -110,7 +110,8 @@ func (sp *SwapPeer) handle(ctx context.Context, msg interface{}) error { var price *big.Int //the message is one which needs accounting... - if _, ok := msg.(PricedMsg); ok { + //only account if swapAccount != nil (== swap is disabled) + if _, ok := msg.(PricedMsg); ok && sp.swapAccount != nil { //..so first check if there are enough funds for the operation available //(for crediting, this means if we are not essentially "overdrafting", or crossing the threshold) price, err = sp.checkAvailableFunds(ctx, msg, CreditEntry) @@ -144,7 +145,8 @@ func (sp *SwapPeer) Send(ctx context.Context, msg interface{}) error { var price *big.Int //the message is one which needs accounting... - if _, ok := msg.(PricedMsg); ok { + //only account if swapAccount != nil (== swap is disabled) + if _, ok := msg.(PricedMsg); ok && sp.swapAccount != nil { //..so first check if there are enough funds for the operation available price, err = sp.checkAvailableFunds(ctx, msg, DebitEntry) //if not (or some other error occured), return error @@ -263,18 +265,21 @@ func (sp *SwapPeer) issueCheque(ctx context.Context) error { //Create a new swap accounted peer func NewSwapPeer(peer *protocols.Peer, swap *Swap) *SwapPeer { - balance := big.NewInt(0) - //check if there is one already in the stateStore and load it - swap.stateStore.Get(peer.String()[:24]+"-swap", &balance) sp := &SwapPeer{ Peer: peer, swapAccount: swap, - balance: balance, storeID: peer.String()[:24] + "-swap", } - swap.lock.Lock() - defer swap.lock.Unlock() - swap.peers[peer.ID()] = sp + //swap is not enabled + if swap != nil { + //check if there is one already in the stateStore and load it + balance := &big.Int{} + swap.stateStore.Get(peer.String()[:24]+"-swap", &balance) + sp.balance = balance + swap.lock.Lock() + defer swap.lock.Unlock() + swap.peers[peer.ID()] = sp + } return sp } diff --git a/swarm/swap_test.go b/swarm/swap_test.go index dfca84197f..69fb702780 100644 --- a/swarm/swap_test.go +++ b/swarm/swap_test.go @@ -22,9 +22,16 @@ import ( "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() @@ -49,6 +56,9 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) { 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 @@ -67,16 +77,19 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) { 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 { 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 { key, data, err := uploadFile(sim.Service("swarm", id).(*Swarm)) if err != nil { @@ -90,6 +103,7 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) { }) } + //wait for kademlia to be healthy if _, err := sim.WaitTillHealthy(ctx, 2); err != nil { return err } @@ -103,8 +117,11 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) { } }) + //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[discover.NodeID]map[discover.NodeID]*big.Int) + //iterate all nodes for _, node := range sim.NodeIDs() { item, ok := sim.NodeItem(node, bucketKeySwarm) if !ok { @@ -113,12 +130,17 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) { } swarm := item.(*Swarm) + //submap for each node is a map of all nodes with the balance for that node subBalances := make(map[discover.NodeID]*big.Int) + //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 := swarm.swap.GetPeerBalance(n) if balance != nil { subBalances[n] = balance @@ -127,9 +149,11 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) { 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())) @@ -139,12 +163,22 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) { } } + //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 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: %s", k.TerminalString(), n.TerminalString(), balanceKwithN)) log.Trace(fmt.Sprintf("balance of %s with %s: %s", n.TerminalString(), k.TerminalString(), mapForSubK[k])) + //...check that they have the same balance in Abs terms and that it is not 0 if balanceKwithN.CmpAbs(mapForSubK[k]) != 0 && balanceKwithN.Cmp(big.NewInt(0)) != 0 { log.Error("Expected balances to be |abs| = 0 AND balance1 != 0, but they are not") } @@ -159,6 +193,8 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) { 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 @@ -185,6 +221,8 @@ func TestSwapNetworkAsymmetricFileUpload(t *testing.T) { } config.Init(privkey) + //enable swap + config.SwapEnabled = true swarm, err := NewSwarm(config, nil) if err != nil { @@ -209,6 +247,9 @@ func TestSwapNetworkAsymmetricFileUpload(t *testing.T) { t.Fatal(err) } + //this is actually quite a big maxFileSize, which results + //in the test running for nearly 2 minutes + //maybe for the test, we could reduce it const maxFileSize = 1024 * 1024 * 4 //1024 bytes * 1024 * 4 = 4MB const minfileSize = 1024 @@ -289,22 +330,21 @@ func TestSwapNetworkAsymmetricFileUpload(t *testing.T) { } /* - TODO: Should balances in this case also be symmetric? - - 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: %s", k.TerminalString(), n.TerminalString(), balanceKwithN)) - log.Trace(fmt.Sprintf("balance of %s with %s: %s", n.TerminalString(), k.TerminalString(), mapForSubK[k])) - if balanceKwithN.CmpAbs(mapForSubK[k]) != 0 && balanceKwithN.Cmp(big.NewInt(0)) != 0 { - log.Error("Expected balances to be |abs| = 0 AND balance1 != 0, but they are not") - } + Assuming that in this case, balances should be symmetric too I + */ + 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: %s", k.TerminalString(), n.TerminalString(), balanceKwithN)) + log.Trace(fmt.Sprintf("balance of %s with %s: %s", n.TerminalString(), k.TerminalString(), mapForSubK[k])) + if balanceKwithN.CmpAbs(mapForSubK[k]) != 0 && balanceKwithN.Cmp(big.NewInt(0)) != 0 { + log.Error("Expected balances to be |abs| = 0 AND balance1 != 0, but they are not") } } } } - */ + } if result.Error != nil { t.Fatal(result.Error) diff --git a/swarm/swarm.go b/swarm/swarm.go index acf29777dd..29ca6bfc89 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -178,6 +178,12 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e if err := nodeID.UnmarshalText([]byte(config.NodeID)); err != nil { return nil, err } + if config.SwapEnabled { + self.swap, err = swap.New(stateStore) + if err != nil { + return nil, err + } + } self.streamer = stream.NewRegistry(nodeID, delivery, self.netStore, stateStore, &stream.RegistryOptions{ SkipCheck: config.DeliverySkipCheck, DoSync: config.SyncEnabled, @@ -344,7 +350,7 @@ func (self *Swarm) Start(srv *p2p.Server) error { newaddr := self.bzz.UpdateLocalAddr([]byte(srv.Self().String())) log.Info("Updated bzz local addr", "oaddr", fmt.Sprintf("%x", newaddr.OAddr), "uaddr", fmt.Sprintf("%s", newaddr.UAddr)) // set chequebook - if self.config.SwapEnabled { + if self.config.SwapEnabled && self.config.SwapAPI != "" { ctx := context.Background() // The initial setup has no deadline. err := self.SetChequebook(ctx) if err != nil { @@ -364,15 +370,6 @@ func (self *Swarm) Start(srv *p2p.Server) error { } log.Info("Swarm network started", "bzzaddr", fmt.Sprintf("%x", self.bzz.Hive.BaseAddr())) - /* - err = self.swap.Start(srv) - if err != nil { - log.Error("swap failed", "err", err) - return err - } - log.Debug("Swap accounting initialized") - */ - if self.ps != nil { self.ps.Start(srv) }