From fac86b399f75c52efd30963f1b00b28340970697 Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Tue, 14 Aug 2018 19:43:25 -0500 Subject: [PATCH] swarm/swap: attach swap to swarm, not registry --- swarm/network/stream/stream.go | 6 +- swarm/network_test.go | 115 ++++++++++++++++++++++++++++++++- swarm/swap/api.go | 7 +- swarm/swap/swap.go | 32 ++++++--- swarm/swap/swap_test.go | 18 ++++++ swarm/swarm.go | 6 -- 6 files changed, 160 insertions(+), 24 deletions(-) diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 7576e91da9..106a4b6a9e 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -104,11 +104,7 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy RegisterSwarmSyncerServer(streamer, syncChunkStore) RegisterSwarmSyncerClient(streamer, syncChunkStore) - var err error - streamer.swap, err = swap.NewSwap(swap.NewDefaultSwapParams().Params) - if err != nil { - log.Error(err.Error()) - } + streamer.swap = swap if options.DoSync { // latestIntC function ensures that diff --git a/swarm/network_test.go b/swarm/network_test.go index aab1c05099..589f687cf7 100644 --- a/swarm/network_test.go +++ b/swarm/network_test.go @@ -40,9 +40,11 @@ import ( ) var ( - loglevel = flag.Int("loglevel", 2, "verbosity of logs") - longrunning = flag.Bool("longrunning", false, "do run long-running tests") - waitKademlia = flag.Bool("waitkademlia", false, "wait for healthy kademlia before checking files availability") + loglevel = flag.Int("loglevel", 2, "verbosity of logs") + longrunning = flag.Bool("longrunning", false, "do run long-running tests") + waitKademlia = flag.Bool("waitkademlia", false, "wait for healthy kademlia before checking files availability") + bucketKeySwap = simulation.BucketKey("swap") + bucketKeySwarm = simulation.BucketKey("swarm") ) func init() { @@ -373,6 +375,113 @@ func testSwarmNetwork(t *testing.T, o *testSwarmNetworkOptions, steps ...testSwa } } +func TestSwapNetworkSymmetricFileUpload(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() + + dir, err := ioutil.TempDir("", "swap-network-test-node") + 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) + + 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) + } + + 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] + }) + for _, id := range nodeIDs { + key, data, err := uploadFile(sim.Service("swarm", id).(*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, + }) + } + + if _, err := sim.WaitTillHealthy(ctx, 2); err != nil { + return err + } + + // 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 { + return nil + } + } + }) + + for _, node := range sim.NodeIDs() { + item, ok := sim.NodeItem(node, bucketKeySwarm) + if !ok { + log.Error("No swarm") + return + } + swarm := item.(*Swarm) + + for _, n := range sim.NodeIDs() { + if node == n { + continue + } + if swarm.swap.GetPeerBalance(n) != nil { + log.Error(fmt.Sprintf("Balance of node %s to node %s: %s", node.TerminalString(), n.TerminalString(), swarm.swap.GetPeerBalance(n).String())) + } else { + log.Error(fmt.Sprintf("Node %s has no balance with node %s", node.TerminalString(), n.TerminalString())) + } + } + } + + 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 uploadFile(swarm *Swarm) (storage.Address, string, error) { diff --git a/swarm/swap/api.go b/swarm/swap/api.go index 632a62433f..c731cb972b 100644 --- a/swarm/swap/api.go +++ b/swarm/swap/api.go @@ -18,9 +18,11 @@ package swap import ( "context" + "math/big" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/p2p/discover" ) // Wrapper for receiving pss messages when using the pss API @@ -32,6 +34,7 @@ type APIMsg struct { // Additional public methods accessible through API for pss type API struct { *SwapProtocol + *Swap } type SwapMetrics struct { @@ -44,8 +47,8 @@ func NewAPI(swap *SwapProtocol) *API { return &API{SwapProtocol: swap} } -func (swapapi *API) Balance(ctx context.Context) (balance int, err error) { - balance = 0 +func (swapapi *API) Balance(ctx context.Context, peer discover.NodeID) (balance *big.Int, err error) { + balance = big.NewInt(0) err = nil return } diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go index b67c53768e..62b8b2bdd0 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -34,6 +34,7 @@ import ( "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/swarm/log" + "github.com/ethereum/go-ethereum/swarm/state" whisper "github.com/ethereum/go-ethereum/whisper/whisperv5" ) @@ -64,9 +65,10 @@ const ( // Swift Automatic Payments // a peer to peer micropayment system type Swap struct { - lock sync.RWMutex - peers map[discover.NodeID]*swapPeer - local *Params // local peer's swap parameters + stateStore state.Store + lock sync.RWMutex + peers map[discover.NodeID]*swapPeer + local *Params // local peer's swap parameters } type EntryDirection bool @@ -83,11 +85,14 @@ type SwapAccountedMsgType interface { func (swap *Swap) AccountForMsg(ctx context.Context, msg interface{}, peer discover.NodeID) error { if accounted, ok := msg.(SwapAccountedMsgType); ok { if _, exists := swap.peers[peer]; !exists { + balance := big.NewInt(0) + swap.stateStore.Get(peer.String()[:24]+"-swap", &balance) swap.lock.Lock() swap.peers[peer] = &swapPeer{ peer: peer, swapAccount: swap, - balance: big.NewInt(0), + balance: balance, + storeID: peer.String()[:24] + "-swap", } swap.lock.Unlock() } @@ -98,6 +103,13 @@ func (swap *Swap) AccountForMsg(ctx context.Context, msg interface{}, peer disco return nil } +func (swap *Swap) GetPeerBalance(peer discover.NodeID) *big.Int { + if p, ok := swap.peers[peer]; ok { + return p.balance + } + return nil +} + // Profile - public swap profile // public parameters for SWAP, serializable config struct passed in handshake type Profile struct { @@ -161,6 +173,7 @@ type swapPeer struct { peer discover.NodeID swapAccount *Swap balance *big.Int + storeID string } func (sp *swapPeer) AccountMsgForPeer(price *big.Int, direction EntryDirection) { @@ -173,21 +186,24 @@ func (sp *swapPeer) AccountMsgForPeer(price *big.Int, direction EntryDirection) } else if direction == DebitEntry { sp.balance = sp.balance.Sub(sp.balance, price) } + //TODO: save to store here? init store? + sp.swapAccount.stateStore.Put(sp.storeID, sp.balance) if sp.balance.Cmp(payAt) > -1 { //TODO: Issue Cheque } if sp.balance.Cmp(dropAt) < 0 { //TODO: Drop peer } - log.Error(fmt.Sprintf("balance for peer %s: %s", sp.peer, sp.balance.String())) + log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.peer, sp.balance.String())) } // New - swap constructor -func NewSwap(local *Params) (swap *Swap, err error) { +func NewSwap(local *Params, stateStore state.Store) (swap *Swap, err error) { swap = &Swap{ - local: local, - peers: make(map[discover.NodeID]*swapPeer), + local: local, + stateStore: stateStore, + peers: make(map[discover.NodeID]*swapPeer), } //swap.SetParams(local) diff --git a/swarm/swap/swap_test.go b/swarm/swap/swap_test.go index 2be0c5013f..060161b897 100644 --- a/swarm/swap/swap_test.go +++ b/swarm/swap/swap_test.go @@ -17,16 +17,28 @@ package swap import ( + "context" + "crypto/rand" "flag" "fmt" + "io/ioutil" "os" "path/filepath" "sync" + "sync/atomic" "testing" + "time" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" + "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/rpc" + "github.com/ethereum/go-ethereum/swarm" + "github.com/ethereum/go-ethereum/swarm/api" + "github.com/ethereum/go-ethereum/swarm/network/simulation" + "github.com/ethereum/go-ethereum/swarm/storage" colorable "github.com/mattn/go-colorable" ) @@ -91,6 +103,12 @@ func TestSwapProtocol(t *testing.T) { }, nil } + streamersvc := func(ctx *node.ServiceContext) (node.Service, erro) { + return &stream.API{ + streamer: NewRegistry, + }, nil + } + // register adds the service to the services the servicenode starts when started err = stack_one.Register(swapsvc) if err != nil { diff --git a/swarm/swarm.go b/swarm/swarm.go index 7c6adb5e74..acf29777dd 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -209,12 +209,6 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e self.bzz = network.NewBzz(bzzconfig, to, stateStore, stream.Spec, self.streamer.Run) - /* - self.swap, err = swap.NewSwap(config.Swap.Params) - if err != nil { - return nil, err - } - */ // Pss = postal service over swarm (devp2p over bzz) self.ps, err = pss.NewPss(to, config.Pss) if err != nil {