cerc-io/plugeth
18
0
Fork 1
Code Issues 4 Pull Requests Packages Projects Releases Wiki Activity

Swarm accounting (#18050)

Browse Source 
* swarm: completed 1st phase of swap accounting

* swarm: swap accounting for swarm with p2p accounting

* swarm/swap: addressed PR comments

* swarm/swap: ignore ErrNotFound on stateStore.Get()

* swarm/swap: GetPeerBalance test; add TODO for chequebook API check

* swarm/network/stream: fix NewRegistry calls with new arguments

* swarm/swap: address @justelad's PR comments
This commit is contained in:
holisticode 2018-11-15 17:41:19 -05:00 committed by Viktor Trón
parent 324027640b
commit ffe2fc3bc4
10 changed files with 353 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

8
swarm/network/stream/delivery_test.go
View File

@ -290,7 +290,7 @@ func TestRequestFromPeers(t *testing.T) {
Peer: protocolsPeer,
}, to)
to.On(peer)
r := NewRegistry(addr.ID(), delivery, nil, nil, nil)
r := NewRegistry(addr.ID(), delivery, nil, nil, nil, nil)
// an empty priorityQueue has to be created to prevent a goroutine being called after the test has finished
sp := &Peer{
@ -331,7 +331,7 @@ func TestRequestFromPeersWithLightNode(t *testing.T) {
Peer: protocolsPeer,
}, to)
to.On(peer)
r := NewRegistry(addr.ID(), delivery, nil, nil, nil)
r := NewRegistry(addr.ID(), delivery, nil, nil, nil, nil)
// an empty priorityQueue has to be created to prevent a goroutine being called after the test has finished
sp := &Peer{
Peer: protocolsPeer,
@ -480,7 +480,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
SkipCheck: skipCheck,
Syncing: SyncingDisabled,
Retrieval: RetrievalEnabled,
})
}, nil)
bucket.Store(bucketKeyRegistry, r)
fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams())
@ -655,7 +655,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
Syncing: SyncingDisabled,
Retrieval: RetrievalDisabled,
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) {
Retrieval: RetrievalDisabled,
Syncing: SyncingRegisterOnly,
SkipCheck: skipCheck,
})
}, nil)
bucket.Store(bucketKeyRegistry, r)
r.RegisterClientFunc(externalStreamName, func(p *Peer, t string, live bool) (Client, error) {

2
swarm/network/stream/snapshot_retrieval_test.go
View File

@ -130,7 +130,7 @@ func retrievalStreamerFunc(ctx *adapters.ServiceContext, bucket *sync.Map) (s no
Retrieval: RetrievalEnabled,
Syncing: SyncingAutoSubscribe,
SyncUpdateDelay: 3 * time.Second,
})
}, nil)
fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams())
bucket.Store(bucketKeyFileStore, fileStore)

4
swarm/network/stream/snapshot_sync_test.go
View File

@ -166,7 +166,7 @@ func streamerFunc(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Servic
Retrieval: RetrievalDisabled,
Syncing: SyncingAutoSubscribe,
SyncUpdateDelay: 3 * time.Second,
})
}, nil)
bucket.Store(bucketKeyRegistry, r)
@ -360,7 +360,7 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int)
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
Retrieval: RetrievalDisabled,
Syncing: SyncingRegisterOnly,
})
}, nil)
bucket.Store(bucketKeyRegistry, r)
fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams())

76
swarm/network/stream/stream.go
View File

@ -21,6 +21,7 @@ import (
"errors"
"fmt"
"math"
"reflect"
"sync"
"time"
@ -87,6 +88,9 @@ type Registry struct {
intervalsStore state.Store
autoRetrieval bool //automatically subscribe to retrieve request stream
maxPeerServers int
spec *protocols.Spec //this protocol's spec
balance protocols.Balance //implements protocols.Balance, for accounting
prices protocols.Prices //implements protocols.Prices, provides prices to accounting
}
// RegistryOptions holds optional values for NewRegistry constructor.
@ -99,7 +103,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, balance protocols.Balance) *Registry {
if options == nil {
options = &RegistryOptions{}
}
@ -119,7 +123,10 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy
intervalsStore: intervalsStore,
autoRetrieval: retrieval,
maxPeerServers: options.MaxPeerServers,
balance: balance,
}
streamer.setupSpec()
streamer.api = NewAPI(streamer)
delivery.getPeer = streamer.getPeer
@ -228,6 +235,17 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy
return streamer
}
//we need to construct a spec instance per node instance
func (r *Registry) setupSpec() {
//first create the "bare" spec
r.createSpec()
//if balance is nil, this node has been started without swap support (swapEnabled flag is false)
if r.balance != nil && !reflect.ValueOf(r.balance).IsNil() {
//swap is enabled, so setup the hook
r.spec.Hook = protocols.NewAccounting(r.balance, r.prices)
}
}
// RegisterClient registers an incoming streamer constructor
func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, string, bool) (Client, error)) {
r.clientMu.Lock()
@ -492,7 +510,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.spec)
bp := network.NewBzzPeer(peer)
np := network.NewPeer(bp, r.delivery.kad)
r.delivery.kad.On(np)
@ -716,35 +734,43 @@ func (c *clientParams) clientCreated() {
close(c.clientCreatedC)
}
// Spec is the spec of the streamer protocol
var Spec = &protocols.Spec{
Name: "stream",
Version: 8,
MaxMsgSize: 10 * 1024 * 1024,
Messages: []interface{}{
UnsubscribeMsg{},
OfferedHashesMsg{},
WantedHashesMsg{},
TakeoverProofMsg{},
SubscribeMsg{},
RetrieveRequestMsg{},
ChunkDeliveryMsgRetrieval{},
SubscribeErrorMsg{},
RequestSubscriptionMsg{},
QuitMsg{},
ChunkDeliveryMsgSyncing{},
},
//GetSpec returns the streamer spec to callers
//This used to be a global variable but for simulations with
//multiple nodes its fields (notably the Hook) would be overwritten
func (r *Registry) GetSpec() *protocols.Spec {
return r.spec
}
func (r *Registry) createSpec() {
// Spec is the spec of the streamer protocol
var spec = &protocols.Spec{
Name: "stream",
Version: 8,
MaxMsgSize: 10 * 1024 * 1024,
Messages: []interface{}{
UnsubscribeMsg{},
OfferedHashesMsg{},
WantedHashesMsg{},
TakeoverProofMsg{},
SubscribeMsg{},
RetrieveRequestMsg{},
ChunkDeliveryMsgRetrieval{},
SubscribeErrorMsg{},
RequestSubscriptionMsg{},
QuitMsg{},
ChunkDeliveryMsgSyncing{},
},
}
r.spec = spec
}
func (r *Registry) Protocols() []p2p.Protocol {
return []p2p.Protocol{
{
Name: Spec.Name,
Version: Spec.Version,
Length: Spec.Length(),
Name: r.spec.Name,
Version: r.spec.Version,
Length: r.spec.Length(),
Run: r.runProtocol,
// NodeInfo: ,
// PeerInfo: ,
},
}
}

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

@ -118,7 +118,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
Retrieval: RetrievalDisabled,
Syncing: SyncingAutoSubscribe,
SkipCheck: skipCheck,
})
}, nil)
fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams())
bucket.Store(bucketKeyFileStore, fileStore)

93
swarm/swap/swap.go Normal file
View File

@ -0,0 +1,93 @@
// 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 swap
import (
"errors"
"fmt"
"strconv"
"sync"
"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/state"
)
// SwAP Swarm Accounting Protocol
// a peer to peer micropayment system
// A node maintains an individual balance with every peer
// Only messages which have a price will be accounted for
type Swap struct {
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
}
// New - swap constructor
func New(stateStore state.Store) (swap *Swap) {
swap = &Swap{
stateStore: stateStore,
balances: make(map[enode.ID]int64),
}
return
}
//Swap implements the protocols.Balance interface
//Add is the (sole) accounting function
func (s *Swap) Add(amount int64, peer *protocols.Peer) (err error) {
s.lock.Lock()
defer s.lock.Unlock()
//load existing balances from the state store
err = s.loadState(peer)
if err != nil && err != state.ErrNotFound {
return
}
//adjust the balance
//if amount is negative, it will decrease, otherwise increase
s.balances[peer.ID()] += amount
//save the new balance to the state store
peerBalance := s.balances[peer.ID()]
err = s.stateStore.Put(peer.ID().String(), &peerBalance)
log.Debug(fmt.Sprintf("balance for peer %s: %s", peer.ID().String(), strconv.FormatInt(peerBalance, 10)))
return err
}
//GetPeerBalance returns the balance for a given peer
func (swap *Swap) GetPeerBalance(peer enode.ID) (int64, error) {
swap.lock.RLock()
defer swap.lock.RUnlock()
if p, ok := swap.balances[peer]; ok {
return p, nil
}
return 0, errors.New("Peer not found")
}
//load balances from the state store (persisted)
func (s *Swap) loadState(peer *protocols.Peer) (err error) {
var peerBalance int64
peerID := peer.ID()
//only load if the current instance doesn't already have this peer's
//balance in memory
if _, ok := s.balances[peerID]; !ok {
err = s.stateStore.Get(peerID.String(), &peerBalance)
s.balances[peerID] = peerBalance
}
return
}

184
swarm/swap/swap_test.go Normal file
View File

@ -0,0 +1,184 @@
// 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 swap
import (
"flag"
"fmt"
"io/ioutil"
mrand "math/rand"
"os"
"testing"
"time"
"github.com/ethereum/go-ethereum/log"
"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/swarm/state"
colorable "github.com/mattn/go-colorable"
)
var (
loglevel = flag.Int("loglevel", 2, "verbosity of logs")
)
func init() {
flag.Parse()
mrand.Seed(time.Now().UnixNano())
log.PrintOrigins(true)
log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(colorable.NewColorableStderr(), log.TerminalFormat(true))))
}
//Test getting a peer's balance
func TestGetPeerBalance(t *testing.T) {
//create a test swap account
swap, testDir := createTestSwap(t)
defer os.RemoveAll(testDir)
//test for correct value
testPeer := newDummyPeer()
swap.balances[testPeer.ID()] = 888
b, err := swap.GetPeerBalance(testPeer.ID())
if err != nil {
t.Fatal(err)
}
if b != 888 {
t.Fatalf("Expected peer's balance to be %d, but is %d", 888, b)
}
//test for inexistent node
id := adapters.RandomNodeConfig().ID
_, err = swap.GetPeerBalance(id)
if err == nil {
t.Fatal("Expected call to fail, but it didn't!")
}
if err.Error() != "Peer not found" {
t.Fatalf("Expected test to fail with %s, but is %s", "Peer not found", err.Error())
}
}
//Test that repeated bookings do correct accounting
func TestRepeatedBookings(t *testing.T) {
//create a test swap account
swap, testDir := createTestSwap(t)
defer os.RemoveAll(testDir)
testPeer := newDummyPeer()
amount := mrand.Intn(100)
cnt := 1 + mrand.Intn(10)
for i := 0; i < cnt; i++ {
swap.Add(int64(amount), testPeer.Peer)
}
expectedBalance := int64(cnt * amount)
realBalance := swap.balances[testPeer.ID()]
if expectedBalance != realBalance {
t.Fatal(fmt.Sprintf("After %d credits of %d, expected balance to be: %d, but is: %d", cnt, amount, expectedBalance, realBalance))
}
testPeer2 := newDummyPeer()
amount = mrand.Intn(100)
cnt = 1 + mrand.Intn(10)
for i := 0; i < cnt; i++ {
swap.Add(0-int64(amount), testPeer2.Peer)
}
expectedBalance = int64(0 - (cnt * amount))
realBalance = swap.balances[testPeer2.ID()]
if expectedBalance != realBalance {
t.Fatal(fmt.Sprintf("After %d debits of %d, expected balance to be: %d, but is: %d", cnt, amount, expectedBalance, realBalance))
}
//mixed debits and credits
amount1 := mrand.Intn(100)
amount2 := mrand.Intn(55)
amount3 := mrand.Intn(999)
swap.Add(int64(amount1), testPeer2.Peer)
swap.Add(int64(0-amount2), testPeer2.Peer)
swap.Add(int64(0-amount3), testPeer2.Peer)
expectedBalance = expectedBalance + int64(amount1-amount2-amount3)
realBalance = swap.balances[testPeer2.ID()]
if expectedBalance != realBalance {
t.Fatal(fmt.Sprintf("After mixed debits and credits, expected balance to be: %d, but is: %d", expectedBalance, realBalance))
}
}
//try restoring a balance from state store
//this is simulated by creating a node,
//assigning it an arbitrary balance,
//then closing the state store.
//Then we re-open the state store and check that
//the balance is still the same
func TestRestoreBalanceFromStateStore(t *testing.T) {
//create a test swap account
swap, testDir := createTestSwap(t)
defer os.RemoveAll(testDir)
testPeer := newDummyPeer()
swap.balances[testPeer.ID()] = -8888
tmpBalance := swap.balances[testPeer.ID()]
swap.stateStore.Put(testPeer.ID().String(), &tmpBalance)
swap.stateStore.Close()
swap.stateStore = nil
stateStore, err := state.NewDBStore(testDir)
if err != nil {
t.Fatal(err)
}
var newBalance int64
stateStore.Get(testPeer.ID().String(), &newBalance)
//compare the balances
if tmpBalance != newBalance {
t.Fatal(fmt.Sprintf("Unexpected balance value after sending cheap message test. Expected balance: %d, balance is: %d",
tmpBalance, newBalance))
}
}
//create a test swap account
//creates a stateStore for persistence and a Swap account
func createTestSwap(t *testing.T) (*Swap, string) {
dir, err := ioutil.TempDir("", "swap_test_store")
if err != nil {
t.Fatal(err)
}
stateStore, err2 := state.NewDBStore(dir)
if err2 != nil {
t.Fatal(err2)
}
swap := New(stateStore)
return swap, dir
}
type dummyPeer struct {
*protocols.Peer
}
//creates a dummy protocols.Peer with dummy MsgReadWriter
func newDummyPeer() *dummyPeer {
id := adapters.RandomNodeConfig().ID
protoPeer := protocols.NewPeer(p2p.NewPeer(id, "testPeer", nil), nil, nil)
dummy := &dummyPeer{
Peer: protoPeer,
}
return dummy
}

18
swarm/swarm.go
View File

@ -51,6 +51,7 @@ import (
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/storage/feed"
"github.com/ethereum/go-ethereum/swarm/storage/mock"
"github.com/ethereum/go-ethereum/swarm/swap"
"github.com/ethereum/go-ethereum/swarm/tracing"
)
@ -78,6 +79,7 @@ type Swarm struct {
netStore *storage.NetStore
sfs *fuse.SwarmFS // need this to cleanup all the active mounts on node exit
ps *pss.Pss
swap *swap.Swap
tracerClose io.Closer
}
@ -171,6 +173,14 @@ 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 {
balancesStore, err := state.NewDBStore(filepath.Join(config.Path, "balances.db"))
if err != nil {
return nil, err
}
self.swap = swap.New(balancesStore)
}
var nodeID enode.ID
if err := nodeID.UnmarshalText([]byte(config.NodeID)); err != nil {
return nil, err
@ -193,7 +203,7 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e
SyncUpdateDelay: config.SyncUpdateDelay,
MaxPeerServers: config.MaxStreamPeerServers,
}
self.streamer = stream.NewRegistry(nodeID, delivery, self.netStore, stateStore, registryOptions)
self.streamer = stream.NewRegistry(nodeID, delivery, self.netStore, stateStore, registryOptions, self.swap)
// Swarm Hash Merklised Chunking for Arbitrary-length Document/File storage
self.fileStore = storage.NewFileStore(self.netStore, self.config.FileStoreParams)
@ -216,7 +226,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)
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)
@ -353,7 +363,9 @@ func (self *Swarm) Start(srv *p2p.Server) error {
newaddr := self.bzz.UpdateLocalAddr([]byte(srv.Self().String()))
log.Info("Updated bzz local addr", "oaddr", fmt.Sprintf("%x", newaddr.OAddr), "uaddr", fmt.Sprintf("%s", newaddr.UAddr))
// set chequebook
if self.config.SwapEnabled {
//TODO: Currently if swap is enabled and no chequebook (or inexistent) contract is provided, the node would crash.
//Once we integrate back the contracts, this check MUST be revisited
if self.config.SwapEnabled && self.config.SwapAPI != "" {
ctx := context.Background() // The initial setup has no deadline.
err := self.SetChequebook(ctx)
if err != nil {