From 4cd40f108f37ad5635f97e48860dd4b30ed074dc Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Thu, 4 Oct 2018 17:58:18 -0500 Subject: [PATCH] swarm/swap: first commit for p2p accounting --- p2p/protocols/accounting.go | 77 ++++ p2p/protocols/protocol.go | 32 +- swarm/network/stream/common_test.go | 2 +- swarm/network/stream/delivery_test.go | 4 +- swarm/network/stream/intervals_test.go | 2 +- swarm/network/stream/snapshot_sync_test.go | 6 +- swarm/network/stream/stream.go | 128 +++++-- swarm/network/stream/syncer_test.go | 2 +- swarm/swap/api.go | 16 +- swarm/swap/chequemanager.go | 8 +- swarm/swap/swap.go | 285 ++++----------- swarm/swap/swap_test.go | 386 +++++++-------------- swarm/swap_test.go | 238 +++++++------ swarm/swarm.go | 12 +- 14 files changed, 564 insertions(+), 634 deletions(-) create mode 100644 p2p/protocols/accounting.go diff --git a/p2p/protocols/accounting.go b/p2p/protocols/accounting.go new file mode 100644 index 0000000000..4c3e95e67b --- /dev/null +++ b/p2p/protocols/accounting.go @@ -0,0 +1,77 @@ +// 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 protocols + +type PriceOracle interface { + Price(uint32, interface{}) (EntryDirection, uint64) + Accountable(interface{}) bool +} + +type BalanceManager interface { + //Credit is crediting the peer, charging local node + Credit(peer *Peer, amount uint64) error + //Debit is crediting the local node, charging the remote peer + Debit(peer *Peer, amount uint64) error +} + +type EntryDirection bool + +const ( + ChargeSender EntryDirection = true + ChargeReceiver EntryDirection = false +) + +type AccountingHook struct { + BalanceManager + PriceOracle +} + +func NewAccountingHook(mgr BalanceManager, po PriceOracle) *AccountingHook { + ah := &AccountingHook{ + PriceOracle: po, + BalanceManager: mgr, + } + return ah +} + +func (ah *AccountingHook) Send(peer *Peer, size uint32, msg interface{}) error { + var err error + if !ah.PriceOracle.Accountable(msg) { + return nil + } + direction, price := ah.PriceOracle.Price(size, msg) + if direction == ChargeSender { + err = ah.BalanceManager.Debit(peer, price) + } else { + err = ah.BalanceManager.Credit(peer, price) + } + return err +} + +func (ah *AccountingHook) Receive(peer *Peer, size uint32, msg interface{}) error { + var err error + if !ah.PriceOracle.Accountable(msg) { + return nil + } + direction, price := ah.PriceOracle.Price(size, msg) + if direction == ChargeReceiver { + err = ah.BalanceManager.Debit(peer, price) + } else { + err = ah.BalanceManager.Credit(peer, price) + } + return err +} diff --git a/p2p/protocols/protocol.go b/p2p/protocols/protocol.go index 59115933fc..b0e1c3a865 100644 --- a/p2p/protocols/protocol.go +++ b/p2p/protocols/protocol.go @@ -122,6 +122,11 @@ type WrappedMsg struct { Payload []byte } +type Hook interface { + Send(*Peer, uint32, interface{}) error + Receive(*Peer, uint32, interface{}) error +} + // Spec is a protocol specification including its name and version as well as // the types of messages which are exchanged type Spec struct { @@ -141,22 +146,13 @@ type Spec struct { // each message must have a single unique data type Messages []interface{} - Services map[string]ProtocolService + Hook Hook initOnce sync.Once codes map[reflect.Type]uint64 types map[uint64]reflect.Type } -type ProtocolService interface { - AddServiceData(data interface{}) - IsSupported(key interface{}) -} - -func (s *Spec) RegisterProtocolService(key string, service ProtocolService) { - s.Services[key] = service -} - func (s *Spec) init() { s.initOnce.Do(func() { s.codes = make(map[reflect.Type]uint64, len(s.Messages)) @@ -289,6 +285,15 @@ func (p *Peer) Send(ctx context.Context, msg interface{}) error { if !found { return errorf(ErrInvalidMsgType, "%v", code) } + + if p.spec.Hook != nil { + err := p.spec.Hook.Send(p, wmsg.Size, msg) + if err != nil { + p.Drop(err) + return err + } + } + return p2p.Send(p.rw, code, wmsg) } @@ -347,6 +352,13 @@ func (p *Peer) handleIncoming(handle func(ctx context.Context, msg interface{}) return errorf(ErrDecode, "<= %v: %v", msg, err) } + if p.spec.Hook != nil { + err := p.spec.Hook.Receive(p, wmsg.Size, val) + if err != nil { + return err + } + } + // call the registered handler callbacks // a registered callback take the decoded message as argument as an interface // which the handler is supposed to cast to the appropriate type diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index 72fdb2bd96..7b536db3fd 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -114,7 +114,7 @@ func newStreamerTester(t *testing.T, registryOptions *RegistryOptions) (*p2ptest delivery := NewDelivery(to, netStore) netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New - streamer := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), registryOptions) + streamer := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), registryOptions, nil) teardown := func() { streamer.Close() removeDataDir() diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index c6ebae3f0e..f4fda4cb49 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -337,7 +337,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{ SkipCheck: skipCheck, - }) + }, nil) bucket.Store(bucketKeyRegistry, r) fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams()) @@ -525,7 +525,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip SkipCheck: skipCheck, DoSync: true, SyncUpdateDelay: 0, - }) + }, nil) fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams()) bucket.Store(bucketKeyFileStore, fileStore) diff --git a/swarm/network/stream/intervals_test.go b/swarm/network/stream/intervals_test.go index 3164193b34..a0af67bf99 100644 --- a/swarm/network/stream/intervals_test.go +++ b/swarm/network/stream/intervals_test.go @@ -84,7 +84,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{ SkipCheck: skipCheck, - }) + }, nil) bucket.Store(bucketKeyRegistry, r) r.RegisterClientFunc(externalStreamName, func(p *Peer, t string, live bool) (Client, error) { diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index 0d58494873..5d42341003 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/swarm/network/stream/snapshot_sync_test.go @@ -163,12 +163,10 @@ func streamerFunc(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Servic kad := network.NewKademlia(addr.Over(), network.NewKadParams()) delivery := NewDelivery(kad, netStore) netStore.NewNetFetcherFunc = network.NewFetcherFactory(dummyRequestFromPeers, true).New - r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{ DoSync: true, SyncUpdateDelay: 3 * time.Second, - }) - + }, nil) bucket.Store(bucketKeyRegistry, r) cleanup = func() { @@ -358,7 +356,7 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int) delivery := NewDelivery(kad, netStore) netStore.NewNetFetcherFunc = network.NewFetcherFactory(dummyRequestFromPeers, true).New - r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), nil) + r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), nil, nil) bucket.Store(bucketKeyRegistry, r) fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams()) diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index fe975f40e4..12ff7120e5 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -61,6 +61,8 @@ type Registry struct { intervalsStore state.Store doRetrieve bool maxPeerServers int + balanceMgr protocols.BalanceManager + priceOracle protocols.PriceOracle } // RegistryOptions holds optional values for NewRegistry constructor. @@ -73,7 +75,7 @@ type RegistryOptions struct { } // NewRegistry is Streamer constructor -func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.SyncChunkStore, intervalsStore state.Store, options *RegistryOptions) *Registry { +func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.SyncChunkStore, intervalsStore state.Store, options *RegistryOptions, balanceMgr protocols.BalanceManager) *Registry { if options == nil { options = &RegistryOptions{} } @@ -180,6 +182,9 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy }() } + streamer.balanceMgr = balanceMgr + streamer.priceOracle = streamer.createPriceOracle() + return streamer } @@ -448,7 +453,7 @@ func (r *Registry) updateSyncing() { } func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error { - peer := protocols.NewPeer(p, rw, Spec) + peer := protocols.NewPeer(p, rw, r.GetSpec()) bp := network.NewBzzPeer(peer) np := network.NewPeer(bp, r.delivery.kad) r.delivery.kad.On(np) @@ -672,32 +677,109 @@ func (c *clientParams) clientCreated() { close(c.clientCreatedC) } -// Spec is the spec of the streamer protocol -var Spec = &protocols.Spec{ - Name: "stream", - Version: 7, - MaxMsgSize: 10 * 1024 * 1024, - Messages: []interface{}{ - UnsubscribeMsg{}, - OfferedHashesMsg{}, - WantedHashesMsg{}, - TakeoverProofMsg{}, - SubscribeMsg{}, - RetrieveRequestMsg{}, - ChunkDeliveryMsgRetrieval{}, - SubscribeErrorMsg{}, - RequestSubscriptionMsg{}, - QuitMsg{}, - ChunkDeliveryMsgSyncing{}, - }, +func (r *Registry) GetSpec() *protocols.Spec { + + // Spec is the spec of the streamer protocol + var spec = &protocols.Spec{ + Name: "stream", + Version: 7, + MaxMsgSize: 10 * 1024 * 1024, + Messages: []interface{}{ + UnsubscribeMsg{}, + OfferedHashesMsg{}, + WantedHashesMsg{}, + TakeoverProofMsg{}, + SubscribeMsg{}, + RetrieveRequestMsg{}, + ChunkDeliveryMsgRetrieval{}, + SubscribeErrorMsg{}, + RequestSubscriptionMsg{}, + QuitMsg{}, + ChunkDeliveryMsgSyncing{}, + }, + Hook: protocols.NewAccountingHook(r.balanceMgr, r.priceOracle), + } + return spec +} + +//An accountable message needs some meta information attached to it +//in order to evaluate the correct price +type priceTag struct { + price uint64 + sizeBased bool + direction protocols.EntryDirection +} +type StreamerPriceOracle struct { + priceMatrix map[uint64]*priceTag + registry *Registry +} + +func (spo *StreamerPriceOracle) Accountable(msg interface{}) bool { + code, ok := spo.registry.GetSpec().GetCode(msg) + if !ok { + return false + } + if _, ok = spo.priceMatrix[code]; ok { + return true + } + return false +} + +func (spo *StreamerPriceOracle) Price(size uint32, msg interface{}) (direction protocols.EntryDirection, price uint64) { + code, ok := spo.registry.GetSpec().GetCode(msg) + if !ok { + panic("Attempting to get message code for an expected message type, but code not found") + } + + tag := spo.priceMatrix[code] + price = tag.price + if tag.sizeBased { + price = price * uint64(size) + } + direction = tag.direction + return +} + +func (r *Registry) createPriceOracle() protocols.PriceOracle { + + po := &StreamerPriceOracle{ + registry: r, + } + po.priceMatrix = make(map[uint64]*priceTag) + + deliveryCode, ok := r.GetSpec().GetCode(ChunkDeliveryMsgRetrieval{}) + if !ok { + panic("Attempting to get message code for an expected message type, but code not found") + } + tag := &priceTag{ + price: uint64(100), + sizeBased: true, + direction: protocols.ChargeReceiver, + } + po.priceMatrix[deliveryCode] = tag + + retrieveReqCode, ok := r.GetSpec().GetCode(RetrieveRequestMsg{}) + if !ok { + panic("Attempting to get message code for an expected message type, but code not found") + } + tag = &priceTag{ + price: uint64(10), + sizeBased: false, + direction: protocols.ChargeSender, + } + po.priceMatrix[retrieveReqCode] = tag + + return po + } func (r *Registry) Protocols() []p2p.Protocol { + spec := r.GetSpec() return []p2p.Protocol{ { - Name: Spec.Name, - Version: Spec.Version, - Length: Spec.Length(), + Name: spec.Name, + Version: spec.Version, + Length: spec.Length(), Run: r.runProtocol, // NodeInfo: , // PeerInfo: , diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index f2be3bef99..eb982c9a1e 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -115,7 +115,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{ SkipCheck: skipCheck, - }) + }, nil) fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams()) bucket.Store(bucketKeyFileStore, fileStore) diff --git a/swarm/swap/api.go b/swarm/swap/api.go index 15ace353dc..efdd94f092 100644 --- a/swarm/swap/api.go +++ b/swarm/swap/api.go @@ -19,7 +19,6 @@ package swap import ( "context" "errors" - "math/big" "github.com/ethereum/go-ethereum/p2p/enode" ) @@ -44,9 +43,10 @@ func NewAPI(swap *Swap) *API { } //Get the balance for this node with a specific peer -func (swapapi *API) BalanceWithPeer(ctx context.Context, peer enode.ID) (balance *big.Int, err error) { - balance = swapapi.swap.peers[peer] - if balance == nil { +func (swapapi *API) BalanceWithPeer(ctx context.Context, peer enode.ID) (balance int64, err error) { + var ok bool + balance, ok = swapapi.swap.balances[peer] + if !ok { err = ErrNoSuchPeerAccounting } return @@ -56,10 +56,10 @@ func (swapapi *API) BalanceWithPeer(ctx context.Context, peer enode.ID) (balance //Iterates over all peers this node is having accounted interaction //and just adds up balances. //It assumes that if a disfavorable balance is represented as a negative value -func (swapapi *API) Balance(ctx context.Context) (balance *big.Int, err error) { - balance = big.NewInt(0) - for _, peerBalance := range swapapi.swap.peers { - balance.Add(balance, peerBalance) +func (swapapi *API) Balance(ctx context.Context) (balance int64, err error) { + balance = 0 + for _, peerBalance := range swapapi.swap.balances { + balance += peerBalance } return } diff --git a/swarm/swap/chequemanager.go b/swarm/swap/chequemanager.go index 1fb2029d72..8e272a6426 100644 --- a/swarm/swap/chequemanager.go +++ b/swarm/swap/chequemanager.go @@ -17,8 +17,6 @@ package swap import ( - "math/big" - "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/swarm/state" ) @@ -33,8 +31,8 @@ type ChequeManager struct { type Cheque struct { serial uint64 timeout uint64 - amount *big.Int - sumCumulated *big.Int + amount int64 + sumCumulated int64 beneficiary enode.ID //this should probably be common.Address? } @@ -48,7 +46,7 @@ func NewChequeManager(stateStore state.Store) *ChequeManager { } } -func (mgr *ChequeManager) CreateCheque(beneficiary enode.ID, amount *big.Int) *Cheque { +func (mgr *ChequeManager) CreateCheque(beneficiary enode.ID, amount int64) *Cheque { mgr.serialPerNode[beneficiary]++ cheque := &Cheque{ serial: mgr.serialPerNode[beneficiary], diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go index 63886d8ffb..fd5a334fb6 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -20,25 +20,26 @@ import ( "context" "errors" "fmt" - "math/big" + "math" + "strconv" "sync" - "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/enode" + "github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/swarm/log" - "github.com/ethereum/go-ethereum/swarm/network/stream" "github.com/ethereum/go-ethereum/swarm/state" ) const ( defaultMaxMsgSize = 1024 * 1024 + OracleID = "swap" swapProtocolName = "swap" swapVersion = 1 ) var ( - payAt = big.NewInt(-4096 * 10000) // threshold that triggers payment {request} (bytes) - dropAt = big.NewInt(-4096 * 12000) // threshold that triggers disconnect (bytes) + payAt = int64(-4096 * 10000) // threshold that triggers payment {request} (bytes) + dropAt = int64(-4096 * 12000) // threshold that triggers disconnect (bytes) ErrNotAccountedMsg = errors.New("Message does not need accounting") ErrInsufficientFunds = errors.New("Insufficient funds") @@ -49,252 +50,116 @@ var ( // A node maintains an individual balance with every peer // Only messages which have a price will be accounted for type Swap struct { - priceOracle PriceOracle //the price oracle facilitates message pricing to the swap instance - chequeManager *ChequeManager //cheque manager keeps track of issued cheques - stateStore state.Store //stateStore is needed in order to keep balances across sessions - lock sync.RWMutex //lock the balances - peers map[enode.ID]*big.Int //map of balances for each peer - protocol *Protocol //reference to the cheque exchange protocol -} - -//PriceOracle is responsible to maintain the price matrix for accounted messages, -//to get its price and to evaluate if a message is accountable or not -type PriceOracle interface { - IsAccountedMsg(event *p2p.PeerEvent) bool - GetPriceForMsg(event *p2p.PeerEvent) (*big.Int, EntryDirection) -} - -//This defines if a price will be debited or credited to an account -type EntryDirection bool - -//For some messages the sender pays (e.g. RetrieveRequestMsg), -//for others the receiver (ChunkDeliveryMsg) -const ( - ChargeSender EntryDirection = true - ChargeReceiver EntryDirection = false -) - -//An accountable message needs some meta information attached to it -//in order to evaluate the correct price -type PriceTag struct { - Price *big.Int - SizeBased bool - Direction EntryDirection -} - -//The default price oracle -type DefaultPriceOracle struct { - priceMatrix map[string]map[uint64]*PriceTag -} - -//Load a default price matrix -func (dpo *DefaultPriceOracle) LoadPriceMatrix() { - dpo.priceMatrix = dpo.loadDefaultPriceMatrix() -} - -//Set up a default price matrix -//This statically builds the price matrix according to previous knowledge -//of which messages need accounting -//The matrix can be queried by -// * the protocol string, returns a sub-map -// * the sub-map by the message code, which returns the PriceTag -func (dpo *DefaultPriceOracle) loadDefaultPriceMatrix() map[string]map[uint64]*PriceTag { - //setup matrix - priceMatrix := make(map[string]map[uint64]*PriceTag) - - //define accounted messages from the stream protocol - streamProtocol := stream.Spec.Name - priceMatrix[streamProtocol] = make(map[uint64]*PriceTag) - - //get indexes of accounted messages in order to populate the map - retrieveRequestMsgIndex, ok := stream.Spec.GetCode(stream.RetrieveRequestMsg{}) - if !ok { - log.Crit("setting up price matrix failed; expected message code not found in Spec!") - return nil - } - deliveryMsgIndex, ok := stream.Spec.GetCode(stream.RetrieveRequestMsg{}) - if !ok { - log.Crit("setting up price matrix failed; expected message code not found in Spec!") - return nil - } - - //assign the price tag to the message - priceMatrix[streamProtocol][uint64(retrieveRequestMsgIndex)] = &PriceTag{ - Price: big.NewInt(10), - SizeBased: false, - Direction: ChargeSender, - } - - priceMatrix[streamProtocol][uint64(deliveryMsgIndex)] = &PriceTag{ - Price: big.NewInt(100), - SizeBased: true, - Direction: ChargeReceiver, - } - - return priceMatrix - -} - -//Check if a message needs accounting -func (dpo *DefaultPriceOracle) IsAccountedMsg(event *p2p.PeerEvent) bool { - var protoPriceMap map[uint64]*PriceTag - var ok bool - if protoPriceMap, ok = dpo.priceMatrix[event.Protocol]; !ok { - return false - } - if _, ok = protoPriceMap[*event.MsgCode]; !ok { - return false - } - return true -} - -//Get the actual price of a message -//If the message is size-based, it returns the calculated price -func (dpo *DefaultPriceOracle) GetPriceForMsg(event *p2p.PeerEvent) (*big.Int, EntryDirection) { - if dpo.IsAccountedMsg(event) { - priceTag := dpo.priceMatrix[event.Protocol][*event.MsgCode] - price := &big.Int{} - price.Set(priceTag.Price) - if priceTag.SizeBased { - price.Mul(priceTag.Price, big.NewInt(int64(*event.MsgSize))) - } - return price, priceTag.Direction - } - return nil, false -} - -//Do the accounting -//Depending on the charging type of the message (set in the PriceTag), -//it will charge the sender or receiver -func (s *Swap) accountMsgForPeer(event *p2p.PeerEvent) { - price, direction := s.priceOracle.GetPriceForMsg(event) - if price == nil { - //TODO what to do in this case? Should not happen - panic("Price is nil; this should have been accounted for but somehow it failed") - } - - if chargeLocal := (direction == ChargeSender) == (event.Type == p2p.PeerEventTypeMsgSend); chargeLocal { - //debit local node and credit remote - s.chargeLocal(event, price) - } else { - //credit local node and debit remote - s.chargeRemote(event, price) - } -} - -//Debit us and credit remote -func (s *Swap) chargeLocal(event *p2p.PeerEvent, amount *big.Int) { - s.lock.Lock() - defer s.lock.Unlock() - //local node is being credited (in its favor), so its balance increases - if s.peers[event.Peer] == nil { - s.peers[event.Peer] = &big.Int{} - s.stateStore.Get(event.Peer.String(), s.peers[event.Peer]) - } - peerBalance := s.peers[event.Peer] - peerBalance.Sub(peerBalance, amount) - //local node is being debited (in favor of remote peer), so its balance decreases - //TODO: save to store here? init store? - s.stateStore.Put(event.Peer.String(), peerBalance) - //(balance *Int) Cmp(payAt) - // -1 if balance < payAt - // 0 if balance == payAt - // +1 if balance > payAt - if peerBalance.Cmp(payAt) == -1 { - ctx := context.TODO() - err := s.issueCheque(ctx, event.Peer) - if err != nil { - //TODO: special error handling, as at this point the accounting has been done - //but the cheque could not be sent? - log.Warn("Payment threshold exceeded, but error sending cheque!", "err", err) - } - } - if peerBalance.Cmp(dropAt) == -1 { - peer := s.protocol.getPeer(event.Peer) - if peer != nil { - peer.Drop(ErrInsufficientFunds) - //this little hack allows for tests to verify that this error occurred - event.Error = ErrInsufficientFunds.Error() - } - } - log.Debug(fmt.Sprintf("balance for peer %s: %s", event.Peer.String(), peerBalance.String())) + chequeManager *ChequeManager //cheque manager keeps track of issued cheques + stateStore state.Store //stateStore is needed in order to keep balances across sessions + lock sync.RWMutex //lock the balances + balances map[enode.ID]int64 //map of balances for each peer + protocol *Protocol //reference to the cheque exchange protocol } //Credit us and debit remote -func (s *Swap) chargeRemote(event *p2p.PeerEvent, amount *big.Int) { +func (s *Swap) Credit(peer *protocols.Peer, amount uint64) (err error) { s.lock.Lock() defer s.lock.Unlock() - //local node is being credited (in its favor), so its balance increases - if s.peers[event.Peer] == nil { - s.peers[event.Peer] = &big.Int{} + + peerBalance := s.balances[peer.ID()] + if _, ok := s.balances[peer.ID()]; !ok { + s.stateStore.Get(peer.ID().String(), &peerBalance) + s.balances[peer.ID()] = peerBalance } - peerBalance := s.peers[event.Peer] - peerBalance.Add(peerBalance, amount) - //local node is being credited(in favor of local peer), so its balance increases - //TODO: save to store here? init store? - s.stateStore.Put(event.Peer.String(), peerBalance) - //(balance *Int) Cmp(payAt) - // -1 if balance < payAt - // 0 if balance == payAt - // +1 if balance > payAt - if peerBalance.Cmp(payAt) == -1 { + //local node is being credited(in favor of local node), so the balance increases + s.balances[peer.ID()] += int64(amount) + peerBalance = s.balances[peer.ID()] + s.stateStore.Put(peer.ID().String(), &peerBalance) + + if float64(peerBalance) > math.Abs(float64(payAt)) { ctx := context.TODO() - err := s.issueCheque(ctx, event.Peer) + err := s.wantCheque(ctx, peer.ID()) if err != nil { //TODO: special error handling, as at this point the accounting has been done //but the cheque could not be sent? log.Warn("Payment threshold exceeded, but error sending cheque!", "err", err) } } - if peerBalance.Cmp(dropAt) == -1 { - peer := s.protocol.getPeer(event.Peer) - if peer != nil { - //this little hack allows for tests to verify that this error occurred - peer.Drop(ErrInsufficientFunds) + if float64(peerBalance) > math.Abs(float64(dropAt)) { + return ErrInsufficientFunds + } + log.Debug(fmt.Sprintf("balance for peer %s: %s", peer.ID().String(), strconv.FormatInt(peerBalance, 10))) + return err +} + +//Debit us and credit remote +func (s *Swap) Debit(peer *protocols.Peer, amount uint64) (err error) { + s.lock.Lock() + defer s.lock.Unlock() + + peerBalance := s.balances[peer.ID()] + if _, ok := s.balances[peer.ID()]; !ok { + s.stateStore.Get(peer.ID().String(), &peerBalance) + s.balances[peer.ID()] = peerBalance + } + //local node is being debited (in favor of remote peer), so its balance decreases + s.balances[peer.ID()] -= int64(amount) + peerBalance = s.balances[peer.ID()] + s.stateStore.Put(peer.ID().String(), &peerBalance) + if peerBalance < payAt { + ctx := context.TODO() + err := s.issueCheque(ctx, peer.ID()) + if err != nil { + //TODO: special error handling, as at this point the accounting has been done + //but the cheque could not be sent? + log.Warn("Payment threshold exceeded, but error sending cheque!", "err", err) } } - log.Debug(fmt.Sprintf("balance for peer %s: %s", event.Peer.String(), peerBalance.String())) + if peerBalance < dropAt { + return ErrInsufficientFunds + } + log.Debug(fmt.Sprintf("balance for peer %s: %s", peer.ID().String(), strconv.FormatInt(peerBalance, 10))) + return nil } //get a peer's balance -func (swap *Swap) GetPeerBalance(peer enode.ID) *big.Int { +func (swap *Swap) GetPeerBalance(peer enode.ID) (int64, error) { swap.lock.RLock() defer swap.lock.RUnlock() - if p, ok := swap.peers[peer]; ok { - return p + if p, ok := swap.balances[peer]; ok { + return p, nil } - return nil + return 0, errors.New("Peer not found") } //Issue a cheque for the remote peer. Happens if we are indebted with the peer //and crossed the payment threshold func (s *Swap) issueCheque(ctx context.Context, id enode.ID) error { - amount := &big.Int{} - cheque := s.chequeManager.CreateCheque(id, amount.Abs(payAt)) - msg := IssueChequeMsg{ + cheque := s.chequeManager.CreateCheque(id, int64(math.Abs(float64(payAt)))) + _ = IssueChequeMsg{ Cheque: cheque, } - //TODO: This should now be via the actual SwapProtocol p := s.protocol.getPeer(id) if p == nil { return fmt.Errorf("wanting to send to non-connected peer!") } - return p.Send(ctx, msg) + return nil + //TODO: Don't actually send any cheques yet + //return p.Send(ctx, msg) +} + +//Issue a cheque for the remote peer. Happens if we are indebted with the peer +//and crossed the payment threshold +func (s *Swap) wantCheque(ctx context.Context, id enode.ID) error { + //TODO: Don't actually initially any real message exchange yet + //return p.Send(ctx, msg) + return nil } // New - swap constructor func New(stateStore state.Store) (swap *Swap) { - priceOracle := &DefaultPriceOracle{} - swap = &Swap{ chequeManager: NewChequeManager(stateStore), stateStore: stateStore, - peers: make(map[enode.ID]*big.Int), - priceOracle: priceOracle, + balances: make(map[enode.ID]int64), protocol: NewProtocol(), } - priceOracle.LoadPriceMatrix() - return } diff --git a/swarm/swap/swap_test.go b/swarm/swap/swap_test.go index c731745342..f2b460d92d 100644 --- a/swarm/swap/swap_test.go +++ b/swarm/swap/swap_test.go @@ -18,10 +18,11 @@ package swap import ( "context" + "crypto/rand" "flag" "fmt" "io/ioutil" - "math/big" + "math" "os" "path/filepath" "testing" @@ -32,6 +33,7 @@ import ( "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" + "github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/swarm/state" colorable "github.com/mattn/go-colorable" @@ -44,6 +46,29 @@ var ( loglevel = flag.Int("loglevel", 2, "verbosity of logs") ) +type testPriceOracle struct{} + +func (tpo *testPriceOracle) Accountable(msg interface{}) bool { + _, ok := testSpec.GetCode(msg) + return ok +} + +func (tpo *testPriceOracle) Price(size uint32, msg interface{}) (protocols.EntryDirection, uint64) { + switch msg := msg.(type) { + case *testExceedsPayAtMsg: + return msg.Price(size) + case *testExceedsDropAtMsg: + return msg.Price(size) + case *testCheapMsg: + return msg.Price(size) + case *testSizeBasedMsg: + return msg.Price(size) + case *testChargeRecvMsg: + return msg.Price(size) + } + return false, 0 +} + var testSpec = &protocols.Spec{ Name: "swapTestSpec", Version: 1, @@ -79,36 +104,34 @@ func (d *dummyRW) ReadMsg() (p2p.Msg, error) { type testExceedsPayAtMsg struct{} type testExceedsDropAtMsg struct{} type testCheapMsg struct{} -type testSizeBasedMsg struct{} +type testSizeBasedMsg struct { + Data []byte +} type testChargeRecvMsg struct{} //this message is just one unit more expensive than the payment threshold -func (tmsg *testExceedsPayAtMsg) Price() *big.Int { - diff := &big.Int{} - diff = diff.Abs(payAt) - return diff.Add(diff, big.NewInt(1)) +func (tmsg *testExceedsPayAtMsg) Price(size uint32) (protocols.EntryDirection, uint64) { + return protocols.ChargeSender, uint64(math.Abs(float64(payAt)) + float64(1)) } //this message is just one unit more expensive than the disconnect threshold -func (tmsg *testExceedsDropAtMsg) Price() *big.Int { - diff := &big.Int{} - diff = diff.Abs(dropAt) - return diff.Add(diff, big.NewInt(1)) +func (tmsg *testExceedsDropAtMsg) Price(size uint32) (protocols.EntryDirection, uint64) { + return protocols.ChargeSender, uint64(math.Abs(float64(dropAt)) + float64(1)) } //a message with an arbitrary cost -func (tmsg *testCheapMsg) Price() *big.Int { - return big.NewInt(100) +func (tmsg *testCheapMsg) Price(size uint32) (protocols.EntryDirection, uint64) { + return protocols.ChargeSender, uint64(10) } //a message which needs to be charged based on price -func (tmsg *testSizeBasedMsg) Price() *big.Int { - return big.NewInt(222) +func (tmsg *testSizeBasedMsg) Price(size uint32) (protocols.EntryDirection, uint64) { + return protocols.ChargeSender, uint64(size) * uint64(100) } //a message which needs to be charged based on price -func (tmsg *testChargeRecvMsg) Price() *big.Int { - return big.NewInt(999) +func (tmsg *testChargeRecvMsg) Price(size uint32) (protocols.EntryDirection, uint64) { + return protocols.ChargeReceiver, uint64(999) } func init() { @@ -120,8 +143,8 @@ func init() { //check that the disconnect threshold is below the payment threshold func TestLimits(t *testing.T) { - if dropAt.Cmp(payAt) > -1 { - t.Fatal(fmt.Sprintf("dropAt limit is not lower than payAt limit, dropAt: %s, payAt: %s", dropAt.String(), payAt.String())) + if dropAt >= payAt { + t.Fatal(fmt.Sprintf("dropAt limit is not lower than payAt limit, dropAt: %d, payAt: %d", dropAt, payAt)) } } @@ -136,32 +159,18 @@ func TestExceedsPayAt(t *testing.T) { swap, testDir := createTestSwap(t) defer os.RemoveAll(testDir) - swap.priceOracle = setupPriceMatrix() + msg := &testExceedsPayAtMsg{} + ctx := context.Background() + testPeer := newDummyPeer() + testPeer.Send(ctx, msg) - node := createAndStartSvcNode(swap, t) - defer node.Stop() - defer os.RemoveAll(node.DataDir()) - - peerID := adapters.RandomNodeConfig().ID - - code, ok := testSpec.GetCode(&testExceedsPayAtMsg{}) - if !ok { - t.Fatal("test message not found in spec") + _, price := msg.Price(0) + if swap.balances[testPeer.ID()] != int64(0-price) { + t.Fatalf("Expected balance to be %d, but is %d", price, swap.balances[testPeer.ID()]) } - event := &p2p.PeerEvent{ - Type: p2p.PeerEventTypeMsgSend, - Protocol: testSpec.Name, - Peer: peerID, - MsgCode: &code, - MsgSize: new(uint32), - Error: "", - } - - swap.handleMsgEvent(event) - - //check that a cheque is present - cheques := swap.chequeManager.openDebitCheques[peerID] + //check that a cheque has been created + cheques := swap.chequeManager.openDebitCheques[testPeer.ID()] if cheques == nil { t.Fatal("Expected cheques for this peer to be present, but are nil") } @@ -171,9 +180,8 @@ func TestExceedsPayAt(t *testing.T) { if cheques[0].serial != 1 { t.Fatal(fmt.Sprintf("Expected the serial to be one (first message but is: %d", cheques[0].serial)) } - absPayAt := big.NewInt(0) - if cheques[0].amount.Cmp(absPayAt.Abs(payAt)) != 0 { - t.Fatal(fmt.Sprintf("Expected the serial to be one (first message but is: %d", cheques[0].serial)) + if float64(cheques[0].amount) != float64(math.Abs(float64(payAt))) { + t.Fatal(fmt.Sprintf("Expected cheques amount to be equal to payAt limit, but it is: %d", cheques[0].amount)) } } @@ -185,79 +193,37 @@ func TestExceedsDropAt(t *testing.T) { swap, testDir := createTestSwap(t) defer os.RemoveAll(testDir) - swap.priceOracle = setupPriceMatrix() + msg := &testExceedsDropAtMsg{} + ctx := context.Background() + testPeer := newDummyPeer() + err := testPeer.Send(ctx, msg) - node := createAndStartSvcNode(swap, t) - defer node.Stop() - defer os.RemoveAll(node.DataDir()) - - //peerID := adapters.RandomNodeConfig().ID - peer := newDummyPeer() - swap.protocol.setPeer(peer.Peer) - - code, ok := testSpec.GetCode(&testExceedsDropAtMsg{}) - if !ok { - t.Fatal("test message not found in spec") + _, price := msg.Price(0) + if swap.balances[testPeer.ID()] != int64(0-price) { + t.Fatalf("Expected balance to be %d, but is %d", price, swap.balances[testPeer.ID()]) } - event := &p2p.PeerEvent{ - Type: p2p.PeerEventTypeMsgSend, - Protocol: testSpec.Name, - Peer: peer.ID(), - MsgCode: &code, - MsgSize: new(uint32), - Error: "", - } - - swap.handleMsgEvent(event) - if event.Error != ErrInsufficientFunds.Error() { - t.Fatal("Excpected this test to fail with insufficient funds, but it did not") + if err != ErrInsufficientFunds { + t.Fatal("Expected this test to fail with insufficient funds, but it did not") } } //send a message with cost, //then check that the balance has the expected amount func TestSendCheapMessage(t *testing.T) { - fmt.Println("send") + //create a test swap account swap, testDir := createTestSwap(t) defer os.RemoveAll(testDir) - swap.priceOracle = setupPriceMatrix() - - node := createAndStartSvcNode(swap, t) - defer node.Stop() - defer os.RemoveAll(node.DataDir()) - - //peerID := adapters.RandomNodeConfig().ID - peer := newDummyPeer() - swap.protocol.setPeer(peer.Peer) - //set an arbitrary test balance value - testBalance := big.NewInt(1234567890) - swap.peers[peer.ID()] = testBalance - - code, ok := testSpec.GetCode(&testCheapMsg{}) - if !ok { - t.Fatal("test message not found in spec") - } - - event := &p2p.PeerEvent{ - Type: p2p.PeerEventTypeMsgSend, - Protocol: testSpec.Name, - Peer: peer.ID(), - MsgCode: &code, - MsgSize: new(uint32), - Error: "", - } - - swap.handleMsgEvent(event) - - peerBalance := swap.peers[peer.ID()] msg := &testCheapMsg{} + ctx := context.Background() + testPeer := newDummyPeer() + testPeer.Send(ctx, msg) //check the new balance - if peerBalance.Cmp(testBalance.Sub(testBalance, msg.Price())) != 0 { - t.Fatal(fmt.Sprintf("Unexpected balance value after sending cheap message test. Expected balance: %s, balance is: %s", - testBalance.Sub(testBalance, msg.Price()).String(), peerBalance.String())) + _, price := msg.Price(0) + if swap.balances[testPeer.ID()] != int64(0-price) { + t.Fatalf("Expected balance to be %d, but is %d", price, swap.balances[testPeer.ID()]) } } @@ -268,48 +234,45 @@ func TestSendCheapMessage(t *testing.T) { //then create a different SwapPeer instance with same peerID, //which will try to load a balance from the stateStore func TestRestoreBalanceFromStateStore(t *testing.T) { + //create a test swap account swap, testDir := createTestSwap(t) defer os.RemoveAll(testDir) - swap.priceOracle = setupPriceMatrix() + msg := &testCheapMsg{} + ctx := context.Background() + testPeer := newDummyPeer() + testPeer.Send(ctx, msg) - node := createAndStartSvcNode(swap, t) - defer node.Stop() - defer os.RemoveAll(node.DataDir()) - - //peerID := adapters.RandomNodeConfig().ID - peer := newDummyPeer() - swap.protocol.setPeer(peer.Peer) - //create a reference an arbitrary balance - testBalance := big.NewInt(1234567890) - //assign the same value to the peer - swap.peers[peer.ID()] = big.NewInt(1234567890) - - code, ok := testSpec.GetCode(&testCheapMsg{}) - if !ok { - t.Fatal("test message not found in spec") + //check the new balance + _, price := msg.Price(0) + expectedBalance := int64(0 - price) + if swap.balances[testPeer.ID()] != expectedBalance { + t.Fatalf("Expected balance to be %d, but is %d", price, swap.balances[testPeer.ID()]) } - event := &p2p.PeerEvent{ - Type: p2p.PeerEventTypeMsgSend, - Protocol: testSpec.Name, - Peer: peer.ID(), - MsgCode: &code, - MsgSize: new(uint32), - Error: "", + var tmpBalance int64 + swap.stateStore.Get(testPeer.ID().String(), &tmpBalance) + //compare the balances + if expectedBalance != tmpBalance { + t.Fatal(fmt.Sprintf("Unexpected balance value in stateStore after sending cheap message test - stateStore should have saved it. Expected balance: %d, balance is: %d", + expectedBalance, tmpBalance)) } - swap.handleMsgEvent(event) + swap.stateStore.Close() + swap.stateStore = nil - peerBalance := swap.peers[peer.ID()] - expectedBalance := &big.Int{} - swap.stateStore.Get(peer.ID().String(), expectedBalance) + stateStore, err := state.NewDBStore(testDir) + if err != nil { + t.Fatal(err) + } + + var newBalance int64 + stateStore.Get(testPeer.ID().String(), &newBalance) //compare the balances - expectedBalance.Sub(testBalance, (&testCheapMsg{}).Price()) - if peerBalance.Cmp(expectedBalance) != 0 { - t.Fatal(fmt.Sprintf("Unexpected balance value after sending cheap message test. Expected balance: %s, balance is: %s", - expectedBalance.String(), peerBalance.String())) + if expectedBalance != newBalance { + t.Fatal(fmt.Sprintf("Unexpected balance value after sending cheap message test. Expected balance: %d, balance is: %d", + expectedBalance, newBalance)) } } @@ -317,49 +280,32 @@ func TestRestoreBalanceFromStateStore(t *testing.T) { //then check that the balance has the expected amount //the message is charged size based func TestSendSizeBasedMsg(t *testing.T) { + //create a test swap account swap, testDir := createTestSwap(t) defer os.RemoveAll(testDir) - swap.priceOracle = setupPriceMatrix() - - node := createAndStartSvcNode(swap, t) - defer node.Stop() - defer os.RemoveAll(node.DataDir()) - - peer := newDummyPeer() - swap.protocol.setPeer(peer.Peer) - //set an arbitrary test balance value - testBalance := big.NewInt(1234567890) - swap.peers[peer.ID()] = testBalance - - code, ok := testSpec.GetCode(&testSizeBasedMsg{}) - if !ok { - t.Fatal("test message not found in spec") - } - - size := new(uint32) - *size = 500 - - event := &p2p.PeerEvent{ - Type: p2p.PeerEventTypeMsgSend, - Protocol: testSpec.Name, - Peer: peer.ID(), - MsgCode: &code, - MsgSize: size, - Error: "", - } - - swap.handleMsgEvent(event) - - peerBalance := swap.peers[peer.ID()] msg := &testSizeBasedMsg{} - expectedBalance := &big.Int{} - expectedBalance.Mul(msg.Price(), big.NewInt(int64(*size))) + msg.Data = make([]byte, 16) + rand.Read(msg.Data) + ctx := context.Background() + testPeer := newDummyPeer() + testPeer.Send(ctx, msg) //check the new balance - if peerBalance.Cmp(testBalance.Sub(testBalance, expectedBalance)) != 0 { - t.Fatal(fmt.Sprintf("Unexpected balance value after sending cheap message test. Expected balance: %s, balance is: %s", - expectedBalance.String(), peerBalance.String())) + r, err := rlp.EncodeToBytes(msg) + if err != nil { + t.Fatal(err) + } + + size := uint32(len(r)) + _, price := msg.Price(size) + expectedBalance := int64(0 - price) + expectedPrice := uint64(size * 100) + if price != expectedPrice { + t.Fatalf("Expected price to be %d, but is %d", expectedPrice, price) + } + if swap.balances[testPeer.ID()] != expectedBalance { + t.Fatalf("Expected balance to be %d, but is %d", expectedBalance, swap.balances[testPeer.ID()]) } } @@ -368,45 +314,19 @@ func TestSendSizeBasedMsg(t *testing.T) { //it means that the balance is changed in positive, //instead of negative as with all other tests func TestChargeAsReceiver(t *testing.T) { + //create a test swap account swap, testDir := createTestSwap(t) defer os.RemoveAll(testDir) - swap.priceOracle = setupPriceMatrix() - - node := createAndStartSvcNode(swap, t) - defer node.Stop() - defer os.RemoveAll(node.DataDir()) - - //peerID := adapters.RandomNodeConfig().ID - peer := newDummyPeer() - swap.protocol.setPeer(peer.Peer) - //set an arbitrary test balance value - testBalance := big.NewInt(1234567890) - swap.peers[peer.ID()] = testBalance - - code, ok := testSpec.GetCode(&testChargeRecvMsg{}) - if !ok { - t.Fatal("test message not found in spec") - } - - event := &p2p.PeerEvent{ - Type: p2p.PeerEventTypeMsgSend, - Protocol: testSpec.Name, - Peer: peer.ID(), - MsgCode: &code, - MsgSize: new(uint32), - Error: "", - } - - swap.handleMsgEvent(event) - - peerBalance := swap.peers[peer.ID()] msg := &testChargeRecvMsg{} + ctx := context.Background() + testPeer := newDummyPeer() + testPeer.Send(ctx, msg) //check the new balance - if peerBalance.Cmp(testBalance.Add(testBalance, msg.Price())) != 0 { - t.Fatal(fmt.Sprintf("Unexpected balance value after sending cheap message test. Expected balance: %s, balance is: %s", - testBalance.Add(testBalance, msg.Price()).String(), peerBalance.String())) + _, price := msg.Price(0) + if swap.balances[testPeer.ID()] != int64(price) { + t.Fatalf("Expected balance to be %d, but is %d", price, swap.balances[testPeer.ID()]) } } @@ -422,44 +342,10 @@ func createTestSwap(t *testing.T) (*Swap, string) { t.Fatal(err2) } swap := New(stateStore) + testSpec.Hook = protocols.NewAccountingHook(swap, &testPriceOracle{}) return swap, dir } -func setupPriceMatrix() PriceOracle { - - priceOracle := &DefaultPriceOracle{} - priceOracle.priceMatrix = make(map[string]map[uint64]*PriceTag) - priceOracle.priceMatrix[testSpec.Name] = make(map[uint64]*PriceTag) - protoMatrix := priceOracle.priceMatrix[testSpec.Name] - protoMatrix[0] = &PriceTag{ - Direction: ChargeSender, - Price: (&testExceedsPayAtMsg{}).Price(), - } - - protoMatrix[1] = &PriceTag{ - Direction: ChargeSender, - Price: (&testExceedsDropAtMsg{}).Price(), - } - - protoMatrix[2] = &PriceTag{ - Direction: ChargeSender, - Price: (&testCheapMsg{}).Price(), - } - - protoMatrix[3] = &PriceTag{ - Direction: ChargeSender, - Price: (&testSizeBasedMsg{}).Price(), - SizeBased: true, - } - - protoMatrix[4] = &PriceTag{ - Direction: ChargeReceiver, - Price: (&testChargeRecvMsg{}).Price(), - } - - return priceOracle -} - //dummy message handler (needed or we will have a panic in the accounting) func dummyMsgHandler(ctx context.Context, msg interface{}) error { return nil @@ -506,14 +392,14 @@ func TestSwapRPC(t *testing.T) { } defer os.RemoveAll(stack.DataDir()) - var balance *big.Int + var balance int64 err = rpcclient.Call(&balance, "swap_balance") if err != nil { t.Fatal("servicenode RPC failed", "err", err) } log.Debug("servicenode balance", "balance", balance) - if balance.Cmp(big.NewInt(0)) != 0 { + if balance != 0 { t.Fatal("Expected balance to be 0 but it is not") } @@ -524,19 +410,17 @@ func TestSwapRPC(t *testing.T) { fake1 := int64(234) fake2 := int64(-100) - fakeBalance1 := big.NewInt(fake1) - fakeBalance2 := big.NewInt(fake2) - swap.peers[id1] = fakeBalance1 - swap.peers[id2] = fakeBalance2 + swap.balances[id1] = fake1 + swap.balances[id2] = fake2 err = rpcclient.Call(&balance, "swap_balanceWithPeer", id1) if err != nil { t.Fatal("servicenode RPC failed", "err", err) } log.Debug("balance1", "balance-1", balance) - if balance.Cmp(fakeBalance1) != 0 { - t.Fatal(fmt.Sprintf("Expected balance %s to be equal to fake balance %s, but it is not", balance.String(), fakeBalance1.String())) + if balance != fake1 { + t.Fatal(fmt.Sprintf("Expected balance %d to be equal to fake balance %d, but it is not", balance, fake1)) } err = rpcclient.Call(&balance, "swap_balanceWithPeer", id2) @@ -544,8 +428,8 @@ func TestSwapRPC(t *testing.T) { t.Fatal("servicenode RPC failed", "err", err) } log.Debug("balance2", "balance-2", balance) - if balance.Cmp(fakeBalance2) != 0 { - t.Fatal(fmt.Sprintf("Expected balance %s to be equal to fake balance %s, but it is not", balance.String(), fakeBalance2.String())) + if balance != fake2 { + t.Fatal(fmt.Sprintf("Expected balance %d to be equal to fake balance %d, but it is not", balance, fake2)) } err = rpcclient.Call(&balance, "swap_balance") @@ -554,9 +438,9 @@ func TestSwapRPC(t *testing.T) { } log.Debug("balance", "balance", balance) - fakeSum := big.NewInt(fake1 + fake2) - if balance.Cmp(fakeSum) != 0 { - t.Fatal(fmt.Sprintf("Expected balance %s to be equal to sum %s, but it is not", balance.String(), fakeSum.String())) + fakeSum := fake1 + fake2 + if balance != fakeSum { + t.Fatal(fmt.Sprintf("Expected balance %d to be equal to sum %d, but it is not", balance, fakeSum)) } } diff --git a/swarm/swap_test.go b/swarm/swap_test.go index fb0c5e4a35..48d3f601bf 100644 --- a/swarm/swap_test.go +++ b/swarm/swap_test.go @@ -18,9 +18,10 @@ package swarm import ( "context" + "errors" "fmt" "io/ioutil" - "math/big" + "math" "math/rand" "os" "strconv" @@ -103,13 +104,24 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) { //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 { - key, data, err := uploadFile(sim.Service("swarm", id).(*Swarm)) + 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 } @@ -121,89 +133,89 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) { }) } - //wait for kademlia to be healthy - 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 + break } } - }) + //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) - //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]*big.Int) - - //iterate all nodes - for _, node := range sim.NodeIDs() { - item, ok := sim.NodeItem(node, bucketKeySwarm) - if !ok { - log.Error("No swarm") - return - } - swarm := item.(*Swarm) - - //submap for each node is a map of all nodes with the balance for that node - subBalances := make(map[enode.ID]*big.Int) - - //iterate all nodes again... - //get all balances with other peers for every node - for _, n := range sim.NodeIDs() { - if node == n { - continue + //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) - //get the peer's balance with this node - balance := swarm.swap.GetPeerBalance(n) - if balance != nil { - subBalances[n] = balance - log.Debug(fmt.Sprintf("Balance of node %s to node %s: %s", node.TerminalString(), n.TerminalString(), swarm.swap.GetPeerBalance(n).String())) - } else { - log.Debug(fmt.Sprintf("Node %s has no balance with node %s", node.TerminalString(), n.TerminalString())) + //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)) + } } } - //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 %s", kk.TerminalString(), vv.String())) - } - } - } + //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 - //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") + //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 math.Abs(float64(balanceKwithN)) != math.Abs(float64(mapForSubK[k])) && balanceKwithN != 0 { + log.Error(fmt.Sprintf("Expected balances to be |abs| = 0 AND balance1 != 0, but they are not: %d, %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) @@ -309,60 +321,62 @@ func TestSwapNetworkAsymmetricFileUpload(t *testing.T) { return nil } } - }) - balancesMap := make(map[enode.ID]map[enode.ID]*big.Int) + 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 - } - swarm := item.(*Swarm) - - subBalances := make(map[enode.ID]*big.Int) - - for _, n := range sim.NodeIDs() { - if node == n { - continue + for _, node := range sim.NodeIDs() { + item, ok := sim.NodeItem(node, bucketKeySwarm) + if !ok { + log.Error("No swarm") + return errors.New("no swarm") } - balance := swarm.swap.GetPeerBalance(n) - if balance != nil { - subBalances[n] = balance - log.Debug(fmt.Sprintf("Balance of node %s to node %s: %s", node.TerminalString(), n.TerminalString(), swarm.swap.GetPeerBalance(n).String())) - } else { - log.Debug(fmt.Sprintf("Node %s has no balance with node %s", node.TerminalString(), n.TerminalString())) + 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)) + } } } - 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 %s", kk.TerminalString(), vv.String())) - } - } - } - - /* - 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") + /* + 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: %d", k.TerminalString(), n.TerminalString(), balanceKwithN)) + log.Trace(fmt.Sprintf("balance of %s with %s: %d", n.TerminalString(), k.TerminalString(), mapForSubK[k])) + if math.Abs(float64(balanceKwithN)) != math.Abs(float64(mapForSubK[k])) && balanceKwithN != 0 { + log.Error("Expected balances to be |abs| = 0 AND balance1 != 0, but they are not") + } } } } } - } + + return nil + }) if result.Error != nil { t.Fatal(result.Error) diff --git a/swarm/swarm.go b/swarm/swarm.go index fd6cbe242c..6fd4cb2bd1 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -174,6 +174,10 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e delivery := stream.NewDelivery(to, self.netStore) self.netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, config.DeliverySkipCheck).New + if config.SwapEnabled { + self.swap = swap.New(stateStore) + } + var nodeID enode.ID if err := nodeID.UnmarshalText([]byte(config.NodeID)); err != nil { return nil, err @@ -184,7 +188,7 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e DoRetrieve: true, SyncUpdateDelay: config.SyncUpdateDelay, MaxPeerServers: config.MaxStreamPeerServers, - }) + }, self.swap) // Swarm Hash Merklised Chunking for Arbitrary-length Document/File storage self.fileStore = storage.NewFileStore(self.netStore, self.config.FileStoreParams) @@ -207,11 +211,7 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e log.Debug("Setup local storage") - self.bzz = network.NewBzz(bzzconfig, to, stateStore, stream.Spec, self.streamer.Run) - - if config.SwapEnabled { - self.swap = swap.New(stateStore) - } + self.bzz = network.NewBzz(bzzconfig, to, stateStore, self.streamer.GetSpec(), self.streamer.Run) // Pss = postal service over swarm (devp2p over bzz) self.ps, err = pss.NewPss(to, config.Pss)