forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 10:22:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

swarm/swap: first commit for p2p accounting

Browse source
This commit is contained in:
Fabio Barone 2018-10-04 17:58:18 -05:00
parent 1ef101b18b
commit 4cd40f108f
14 changed files with 564 additions and 634 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

77
p2p/protocols/accounting.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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
}

32
p2p/protocols/protocol.go
View file

@ -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

2
swarm/network/stream/common_test.go
View file

@ -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()

4
swarm/network/stream/delivery_test.go
View file

@ -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)

2
swarm/network/stream/intervals_test.go
View file

@ -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) {

6
swarm/network/stream/snapshot_sync_test.go
View file

@ -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())

128
swarm/network/stream/stream.go
View file

@ -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: ,

2
swarm/network/stream/syncer_test.go
View file

@ -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)

16
swarm/swap/api.go
View file

@ -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
}

8
swarm/swap/chequemanager.go
View file

@ -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],

285
swarm/swap/swap.go
View file

@ -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
}

386
swarm/swap/swap_test.go
View file

@ -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))
}
}

238
swarm/swap_test.go
View file

@ -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)

12
swarm/swarm.go
View file

@ -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)