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

Merge a222ed4fa0 into 6566a0a3b8

Browse source
This commit is contained in:
holisticode 2018-10-12 19:54:25 +00:00 committed by GitHub
commit ed0fc101fc
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
21 changed files with 1534 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
metrics/counter.go
View file

@ -1,6 +1,8 @@
package metrics
import "sync/atomic"
import (
"sync/atomic"
)
// Counters hold an int64 value that can be incremented and decremented.
type Counter interface {
@ -20,6 +22,17 @@ func GetOrRegisterCounter(name string, r Registry) Counter {
return r.GetOrRegister(name, NewCounter).(Counter)
}
// GetOrRegisterCounterForced returns an existing Counter or constructs and registers a
// new Counter no matter the global switch is enabled or not.
// Be sure to unregister the counter from the registry once it is of no use to
// allow for garbage collection.
func GetOrRegisterCounterForced(name string, r Registry) Counter {
if nil == r {
r = DefaultRegistry
}
return r.GetOrRegister(name, NewCounterForced).(Counter)
}
// NewCounter constructs a new StandardCounter.
func NewCounter() Counter {
if !Enabled {
@ -28,6 +41,12 @@ func NewCounter() Counter {
return &StandardCounter{0}
}
// NewCounterForced constructs a new StandardCounter and returns it no matter if
// the global switch is enabled or not.
func NewCounterForced() Counter {
return &StandardCounter{0}
}
// NewRegisteredCounter constructs and registers a new StandardCounter.
func NewRegisteredCounter(name string, r Registry) Counter {
c := NewCounter()
@ -38,6 +57,19 @@ func NewRegisteredCounter(name string, r Registry) Counter {
return c
}
// NewRegisteredCounterForced constructs and registers a new StandardCounter
// and launches a goroutine no matter the global switch is enabled or not.
// Be sure to unregister the counter from the registry once it is of no use to
// allow for garbage collection.
func NewRegisteredCounterForced(name string, r Registry) Counter {
c := NewCounterForced()
if nil == r {
r = DefaultRegistry
}
r.Register(name, c)
return c
}
// CounterSnapshot is a read-only copy of another Counter.
type CounterSnapshot int64

4
p2p/peer.go
View file

@ -125,6 +125,10 @@ func NewPeer(id enode.ID, name string, caps []Cap) *Peer {
return peer
}
func (p *Peer) Events() *event.Feed {
return p.events
}
// ID returns the node's public key.
func (p *Peer) ID() enode.ID {
return p.rw.node.ID()

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

@ -0,0 +1,159 @@
// 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
import (
"github.com/ethereum/go-ethereum/metrics"
)
//define some metrics
var (
//NOTE: these metrics just define the interfaces and are currently *NOT persisted* over sessions
//All metrics are cumulative
//total amount of units credited
mBalanceCredit = metrics.NewRegisteredCounterForced("account.balance.credit", nil)
//total amount of units debited
mBalanceDebit = metrics.NewRegisteredCounterForced("account.balance.debit", nil)
//total amount of bytes credited
mBytesCredit = metrics.NewRegisteredCounterForced("account.bytes.credit", nil)
//total amount of bytes debited
mBytesDebit = metrics.NewRegisteredCounterForced("account.bytes.debit", nil)
//total amount of credited messages
mMsgCredit = metrics.NewRegisteredCounterForced("account.msg.credit", nil)
//total amount of debited messages
mMsgDebit = metrics.NewRegisteredCounterForced("account.msg.debit", nil)
//how many times local node had to drop remote peers
mPeerDrops = metrics.NewRegisteredCounterForced("account.peerdrops", nil)
//how many times local node overdrafted and dropped
mSelfDrops = metrics.NewRegisteredCounterForced("account.selfdrops", nil)
//how many cheques have been issued
//mChequesIssued = metrics.NewRegisteredCounterForced("account.cheques.issued", nil)
//how many cheques have been received
//mChequesReceived = metrics.NewRegisteredCounterForced("account.cheques.received", nil)
)
//Prices defines how prices are being passed on to the accounting instance
type Prices interface {
//Return the Price for a message
Price(interface{}) *Price
}
//Price represents the costs of a message
type Price struct {
Value int64 //Positive if sender pays, negative if receiver pays
PerByte bool //True if the price is per byte or for unit
}
//Price can be different depending on if it is the sender or receiver who pays
//ForSender gives back the price for sending a message
func (p *Price) ForSender(size uint32) int64 {
price := p.Value
if p.PerByte {
price *= int64(size)
}
return price
}
//Price can be different depending on if it is the sender or receiver who pays
//ForReceiver gives back the price for receiving a message
func (p *Price) ForReceiver(size uint32) int64 {
price := p.Value
if p.PerByte {
price *= int64(size)
}
return 0 - price
}
//Balance is the actual accounting instance
//Balance defines the operations needed for accounting
//Implementations internally maintain the balance for every peer
type Balance interface {
//Adds amount to the local balance with remote node `peer`;
//positive amount = credit local node
//negative amount = debit local node
Add(amount int64, peer *Peer) error
}
//Accounting implements the Hook interface
//It interfaces to the balances through the Balance interface,
//while interfacing with protocols and its prices through the Prices interface
type Accounting struct {
Balance //interface to accounting logic
Prices //interface to prices logic
}
func NewAccounting(mgr Balance, po Prices) *Accounting {
ah := &Accounting{
Prices: po,
Balance: mgr,
}
return ah
}
//Implement Hook.Send
func (ah *Accounting) Send(peer *Peer, size uint32, msg interface{}) error {
//get the price for a message (through the protocol spec)
price := ah.Price(msg)
//this message doesn't need accounting
if price == nil {
return nil
}
//evaluate the price for sending messages
finalPrice := price.ForSender(size)
//do the accounting
err := ah.Add(finalPrice, peer)
//record metrics
ah.doMetrics(finalPrice, size, err)
return err
}
//Implement Hook.Receive
func (ah *Accounting) Receive(peer *Peer, size uint32, msg interface{}) error {
//get the price for a message (through the protocol spec)
price := ah.Price(msg)
//this message doesn't need accounting
if price == nil {
return nil
}
//evaluate the price for receiving messages
finalPrice := price.ForReceiver(size)
//do the accounting
err := ah.Add(finalPrice, peer)
//record metrics
ah.doMetrics(finalPrice, size, err)
return err
}
//record some metrics
func (ah *Accounting) doMetrics(price int64, size uint32, err error) {
if price > 0 {
mBalanceCredit.Inc(int64(price))
mBytesCredit.Inc(int64(size))
mMsgCredit.Inc(1)
if err != nil {
mPeerDrops.Inc(1)
}
} else {
mBalanceDebit.Inc(int64(price))
mBytesDebit.Inc(int64(size))
mMsgDebit.Inc(1)
if err != nil {
mSelfDrops.Inc(1)
}
}
}

329
p2p/protocols/accounting_test.go Normal file
View file

@ -0,0 +1,329 @@
// 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
import (
"bytes"
"context"
"testing"
"time"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/rlp"
)
//dummy Balance implementation
type dummyBalance struct {
amount int64
peer *Peer
}
//dummy Prices implementation
type dummyPrices struct{}
//a dummy message which needs size based accounting
type perBytesMsg struct {
Content string
}
//a dummy message which is paid for per unit
type perUnitMsg struct{}
//a dummy message which doesn't need any payment
type zeroMsg struct{}
//return the price for the defined messages
func (d *dummyPrices) Price(msg interface{}) *Price {
switch msg.(type) {
//size based message cost, sender pays
case *perBytesMsg:
return &Price{
PerByte: true,
Value: int64(100),
}
//unitary cost, receiver pays
case *perUnitMsg:
return &Price{
PerByte: false,
Value: int64(-99),
}
}
return nil
}
//dummy accounting implementation, only stores values for later check
func (d *dummyBalance) Add(amount int64, peer *Peer) error {
d.amount = amount
d.peer = peer
return nil
}
//We need to test that if the hook is not defined, then message infrastructure
//(send,receive) still works
func TestNoHook(t *testing.T) {
//create a test spec
spec := createTestSpec()
//a random node
id := adapters.RandomNodeConfig().ID
//a peer
p := p2p.NewPeer(id, "testPeer", nil)
rw := &dummyRW{}
peer := NewPeer(p, rw, spec)
ctx := context.TODO()
msg := &perBytesMsg{Content: "testBalance"}
//send a message
peer.Send(ctx, msg)
//simulate receiving a message
rw.msg = msg
peer.handleIncoming(func(ctx context.Context, msg interface{}) error {
return nil
})
//all should just work and not result in any error
}
//lowest level unit test
func TestSend(t *testing.T) {
//create instances
balance := &dummyBalance{}
prices := &dummyPrices{}
//create the spec
spec := createTestSpec()
//create the accounting hook for the spec
spec.Hook = NewAccounting(balance, prices)
//create a peer
id := adapters.RandomNodeConfig().ID
p := p2p.NewPeer(id, "testPeer", nil)
peer := NewPeer(p, &dummyRW{}, spec)
msg := &perBytesMsg{Content: "testBalance"}
size, _ := rlp.EncodeToBytes(msg)
spec.Hook.Send(peer, uint32(len(size)), msg)
if balance.amount != int64(len(size)*100) {
t.Fatalf("Expected balance to be %d but is %d", (len(size) * 100), balance.amount)
}
//send a unitary, receiver pays message
msg2 := &perUnitMsg{}
spec.Hook.Send(peer, 0, msg2)
//check that the balance is actually the expected
if balance.amount != int64(-99) {
t.Fatalf("Expected balance to be %d but is %d", -99, balance.amount)
}
//set arbitrary balance
balance.amount = 77
//send a non accounted message, balance should not change
msg3 := &zeroMsg{}
spec.Hook.Send(peer, 0, msg3)
if balance.amount != int64(77) {
t.Fatalf("Expected balance to be %d but is %d", 77, balance.amount)
}
}
//lowest level unit test
func TestReceive(t *testing.T) {
//create instances
balance := &dummyBalance{}
prices := &dummyPrices{}
//create the spec
spec := createTestSpec()
//create the accounting hook for the spec
spec.Hook = NewAccounting(balance, prices)
//create a peer
id := adapters.RandomNodeConfig().ID
p := p2p.NewPeer(id, "testPeer", nil)
peer := NewPeer(p, &dummyRW{}, spec)
msg := &perBytesMsg{Content: "testBalance"}
size, _ := rlp.EncodeToBytes(msg)
spec.Hook.Receive(peer, uint32(len(size)), msg)
if balance.amount != int64(len(size)*-100) {
t.Fatalf("Expected balance to be %d but is %d", (len(size) * 100), balance.amount)
}
//send a unitary, receiver pays message
msg2 := &perUnitMsg{}
spec.Hook.Receive(peer, 0, msg2)
//check that the balance is actually the expected
if balance.amount != int64(99) {
t.Fatalf("Expected balance to be %d but is %d", -99, balance.amount)
}
//set arbitrary balance
balance.amount = 77
//send a non accounted message, balance should not change
msg3 := &zeroMsg{}
spec.Hook.Receive(peer, 0, msg3)
if balance.amount != int64(77) {
t.Fatalf("Expected balance to be %d but is %d", 77, balance.amount)
}
}
//Test sending with a peer, higher level
func TestSendWithPeer(t *testing.T) {
//create instances
balance := &dummyBalance{}
prices := &dummyPrices{}
//create the spec
spec := createTestSpec()
//create the accounting hook for the spec
spec.Hook = NewAccounting(balance, prices)
//create a peer
id := adapters.RandomNodeConfig().ID
p := p2p.NewPeer(id, "testPeer", nil)
peer := NewPeer(p, &dummyRW{}, spec)
ctx := context.TODO()
msg := &perBytesMsg{Content: "testBalance"}
size, _ := rlp.EncodeToBytes(msg)
//send a size based message
peer.Send(ctx, msg)
//check that the balance is actually the expected
if balance.amount != int64((len(size) * 100)) {
t.Fatalf("Expected balance to be %d but is %d", (len(size) * 100), balance.amount)
}
//send a unitary, receiver pays message
msg2 := &perUnitMsg{}
peer.Send(ctx, msg2)
//check that the balance is actually the expected
if balance.amount != int64(-99) {
t.Fatalf("Expected balance to be %d but is %d", -99, balance.amount)
}
//set arbitrary balance
balance.amount = 77
//send a non accounted message, balance should not change
msg3 := &zeroMsg{}
peer.Send(ctx, msg3)
if balance.amount != int64(77) {
t.Fatalf("Expected balance to be %d but is %d", 77, balance.amount)
}
}
//Test receiving with a peer, higher level
func TestReceiveWithPeer(t *testing.T) {
//create instances
balance := &dummyBalance{}
prices := &dummyPrices{}
//create the spec
spec := createTestSpec()
//create the accounting hook for the spec
spec.Hook = NewAccounting(balance, prices)
//create a peer
id := adapters.RandomNodeConfig().ID
p := p2p.NewPeer(id, "testPeer", nil)
rw := &dummyRW{}
peer := NewPeer(p, rw, spec)
//simulate receiving a size based, sender-pays message
msg := &perBytesMsg{Content: "testBalance"}
size, _ := rlp.EncodeToBytes(msg)
rw.msg = msg
rw.code, _ = spec.GetCode(msg)
err := peer.handleIncoming(func(ctx context.Context, msg interface{}) error {
return nil
})
if err != nil {
t.Fatalf("Expected no error, but got error: %v", err)
}
if balance.amount != int64((len(size) * (-100))) {
t.Fatalf("Expected balance to be %d but is %d", (len(size) * (-100)), balance.amount)
}
//simulate receiving a size based, sender-pays message
msg2 := &perUnitMsg{}
rw.msg = msg2
rw.code, _ = spec.GetCode(msg2)
err = peer.handleIncoming(func(ctx context.Context, msg interface{}) error {
return nil
})
if err != nil {
t.Fatalf("Expected no error, but got error: %v", err)
}
if balance.amount != int64(99) {
t.Fatalf("Expected balance to be %d but is %d", 99, balance.amount)
}
//set arbitrary balance an
msg3 := &zeroMsg{}
rw.msg = msg3
rw.code, _ = spec.GetCode(msg3)
//need to reset cause no accounting won't overwrite
balance.amount = -888
//simulate receiving a non accounted message, balance should not change
err = peer.handleIncoming(func(ctx context.Context, msg interface{}) error {
return nil
})
if err != nil {
t.Fatalf("Expected no error, but got error: %v", err)
}
if balance.amount != int64(-888) {
t.Fatalf("Expected balance to be %d but is %d", -888, balance.amount)
}
}
//dummy implementation of a MsgReadWriter
//this allows for quick and easy unit tests without
//having to build up the complete protocol
type dummyRW struct {
msg interface{}
size uint32
code uint64
}
func (d *dummyRW) WriteMsg(msg p2p.Msg) error {
return nil
}
func (d *dummyRW) ReadMsg() (p2p.Msg, error) {
enc := bytes.NewReader(d.getDummyMsg())
return p2p.Msg{
Code: d.code,
Size: d.size,
Payload: enc,
ReceivedAt: time.Now(),
}, nil
}
func (d *dummyRW) getDummyMsg() []byte {
r, _ := rlp.EncodeToBytes(d.msg)
var b bytes.Buffer
wmsg := WrappedMsg{
Context: b.Bytes(),
Size: uint32(len(r)),
Payload: r,
}
rr, _ := rlp.EncodeToBytes(wmsg)
d.size = uint32(len(rr))
return rr
}
//create a test spec
func createTestSpec() *Spec {
spec := &Spec{
Name: "test",
Version: 42,
MaxMsgSize: 10 * 1024,
Messages: []interface{}{
perBytesMsg{},
perUnitMsg{},
zeroMsg{},
},
}
return spec
}

37
p2p/protocols/protocol.go
View file

@ -32,6 +32,7 @@ import (
"bufio"
"bytes"
"context"
"errors"
"fmt"
"io"
"reflect"
@ -122,6 +123,16 @@ type WrappedMsg struct {
Payload []byte
}
//For accounting, the design is to allow the Spec to describe which and how its messages are priced
//To access this functionality, we provide a Hook interface which will call accounting methods
//NOTE: there could be more such (horizontal) hooks in the future
type Hook interface {
//A hook for sending messages
Send(peer *Peer, size uint32, msg interface{}) error
//A hook for receiving messages
Receive(peer *Peer, size uint32, msg 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,6 +152,9 @@ type Spec struct {
// each message must have a single unique data type
Messages []interface{}
//hook for accounting (could be extended to multiple hooks in the future)
Hook Hook
initOnce sync.Once
codes map[reflect.Type]uint64
types map[uint64]reflect.Type
@ -278,6 +292,16 @@ func (p *Peer) Send(ctx context.Context, msg interface{}) error {
if !found {
return errorf(ErrInvalidMsgType, "%v", code)
}
//if the accounting hook is set, call it
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)
}
@ -336,6 +360,19 @@ func (p *Peer) handleIncoming(handle func(ctx context.Context, msg interface{})
return errorf(ErrDecode, "<= %v: %v", msg, err)
}
//if the accounting hook is set, call it
if p.spec.Hook != nil {
if wmsg.Size != uint32(len(wmsg.Payload)) {
errMsg := "Advertised message size and payload length don't match"
log.Warn(errMsg)
p.Drop(errors.New(errMsg))
}
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

133
p2p/protocols/protocol_test.go
View file

@ -197,6 +197,139 @@ func TestProtoHandshakeSuccess(t *testing.T) {
runProtoHandshake(t, &protoHandshake{42, "420"})
}
type dummyHook struct {
peer *Peer
size uint32
msg interface{}
send bool
err error
waitC chan struct{}
}
type dummyMsg struct {
Content string
}
func (d *dummyHook) Send(peer *Peer, size uint32, msg interface{}) error {
d.peer = peer
d.size = size
d.msg = msg
d.send = true
return d.err
}
func (d *dummyHook) Receive(peer *Peer, size uint32, msg interface{}) error {
d.peer = peer
d.size = size
d.msg = msg
d.send = false
d.waitC <- struct{}{}
return d.err
}
func TestProtocolHook(t *testing.T) {
testHook := &dummyHook{
waitC: make(chan struct{}, 1),
}
spec := &Spec{
Name: "test",
Version: 42,
MaxMsgSize: 10 * 1024,
Messages: []interface{}{
dummyMsg{},
},
Hook: testHook,
}
runFunc := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
peer := NewPeer(p, rw, spec)
ctx := context.TODO()
peer.Send(ctx, &dummyMsg{
Content: "handshake"})
handle := func(ctx context.Context, msg interface{}) error {
return nil
}
return peer.Run(handle)
}
conf := adapters.RandomNodeConfig()
tester := p2ptest.NewProtocolTester(t, conf.ID, 2, runFunc)
err := tester.TestExchanges(p2ptest.Exchange{
Expects: []p2ptest.Expect{
{
Code: 0,
Msg: &dummyMsg{Content: "handshake"},
Peer: tester.Nodes[0].ID(),
},
},
})
if err != nil {
t.Fatal(err)
}
if testHook.msg == nil || testHook.msg.(*dummyMsg).Content != "handshake" {
t.Fatal("Expected msg to be set, but it is not")
}
if testHook.send != true {
t.Fatal("Expected a send message, but it is not")
}
if testHook.peer == nil || testHook.peer.ID() != tester.Nodes[0].ID() {
t.Fatal("Expected peer ID to be set correctly, but it is not")
}
if testHook.size != 11 { //11 is the length of the encoded message
t.Fatalf("Expected size to be %d, but it is %d ", 1, testHook.size)
}
err = tester.TestExchanges(p2ptest.Exchange{
Triggers: []p2ptest.Trigger{
{
Code: 0,
Msg: &dummyMsg{Content: "response"},
Peer: tester.Nodes[1].ID(),
},
},
})
<-testHook.waitC
if err != nil {
t.Fatal(err)
}
if testHook.msg == nil || testHook.msg.(*dummyMsg).Content != "response" {
t.Fatal("Expected msg to be set, but it is not")
}
if testHook.send != false {
t.Fatal("Expected a send message, but it is not")
}
if testHook.peer == nil || testHook.peer.ID() != tester.Nodes[1].ID() {
t.Fatal("Expected peer ID to be set correctly, but it is not")
}
if testHook.size != 10 { //11 is the length of the encoded message
t.Fatalf("Expected size to be %d, but it is %d ", 1, testHook.size)
}
testHook.err = fmt.Errorf("dummy error")
err = tester.TestExchanges(p2ptest.Exchange{
Triggers: []p2ptest.Trigger{
{
Code: 0,
Msg: &dummyMsg{Content: "response"},
Peer: tester.Nodes[1].ID(),
},
},
})
<-testHook.waitC
time.Sleep(100 * time.Millisecond)
err = tester.TestDisconnected(&p2ptest.Disconnect{tester.Nodes[1].ID(), testHook.err})
if err != nil {
t.Fatalf("Expected a specific disconnect error, but got different one: %v", err)
}
}
func moduleHandshakeExchange(id enode.ID, resp uint) []p2ptest.Exchange {
return []p2ptest.Exchange{

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

20
swarm/network/stream/delivery.go
View file

@ -173,7 +173,8 @@ func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *
return
}
if req.SkipCheck {
err = sp.Deliver(ctx, chunk, s.priority)
syncing := false
err = sp.Deliver(ctx, chunk, s.priority, syncing)
if err != nil {
log.Warn("ERROR in handleRetrieveRequestMsg", "err", err)
}
@ -189,12 +190,23 @@ func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *
return nil
}
//Chunk delivery always uses the same message type....
type ChunkDeliveryMsg struct {
Addr storage.Address
SData []byte // the stored chunk Data (incl size)
peer *Peer // set in handleChunkDeliveryMsg
Addr storage.Address
SData []byte // the stored chunk Data (incl size)
peer *Peer // set in handleChunkDeliveryMsg
Syncing bool // if true, this is a delivery for syncing (no SWAP accounting needed)
}
//...but swap accounting needs to disambiguate if it is a delivery for syncing or for retrieval
//as it decides based on message type if it needs to account for this message or not
//defines a chunk delivery for retrieval (with accounting)
type ChunkDeliveryMsgRetrieval ChunkDeliveryMsg
//defines a chunk delivery for syncing (without accounting)
type ChunkDeliveryMsgSyncing ChunkDeliveryMsg
// TODO: Fix context SNAFU
func (d *Delivery) handleChunkDeliveryMsg(ctx context.Context, sp *Peer, req *ChunkDeliveryMsg) error {
var osp opentracing.Span

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

3
swarm/network/stream/messages.go
View file

@ -347,7 +347,8 @@ func (p *Peer) handleWantedHashesMsg(ctx context.Context, req *WantedHashesMsg)
return fmt.Errorf("handleWantedHashesMsg get data %x: %v", hash, err)
}
chunk := storage.NewChunk(hash, data)
if err := p.Deliver(ctx, chunk, s.priority); err != nil {
syncing := true
if err := p.Deliver(ctx, chunk, s.priority, syncing); err != nil {
return err
}
}

23
swarm/network/stream/peer.go
View file

@ -128,16 +128,31 @@ func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer {
}
// Deliver sends a storeRequestMsg protocol message to the peer
func (p *Peer) Deliver(ctx context.Context, chunk storage.Chunk, priority uint8) error {
// Depending on the `syncing` parameter we send different message types
func (p *Peer) Deliver(ctx context.Context, chunk storage.Chunk, priority uint8, syncing bool) error {
var sp opentracing.Span
ctx, sp = spancontext.StartSpan(
ctx,
"send.chunk.delivery")
defer sp.Finish()
msg := &ChunkDeliveryMsg{
Addr: chunk.Address(),
SData: chunk.Data(),
var msg interface{}
//we send different types of messages if delivery is for syncing or retrievals,
//even if handling and content of the message are the same,
//because swap accounting decides which messages need accounting based on the message type
if syncing {
msg = &ChunkDeliveryMsgSyncing{
Addr: chunk.Address(),
SData: chunk.Data(),
Syncing: syncing,
}
} else {
msg = &ChunkDeliveryMsgRetrieval{
Addr: chunk.Address(),
SData: chunk.Data(),
Syncing: syncing,
}
}
return p.SendPriority(ctx, msg, priority)
}

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

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

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

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

@ -20,6 +20,7 @@ import (
"context"
"fmt"
"math"
"reflect"
"sync"
"time"
@ -61,6 +62,9 @@ type Registry struct {
intervalsStore state.Store
doRetrieve bool
maxPeerServers int
balanceMgr protocols.Balance
prices protocols.Prices
spec *protocols.Spec
}
// RegistryOptions holds optional values for NewRegistry constructor.
@ -73,13 +77,14 @@ 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.Balance) *Registry {
if options == nil {
options = &RegistryOptions{}
}
if options.SyncUpdateDelay <= 0 {
options.SyncUpdateDelay = 15 * time.Second
}
streamer := &Registry{
addr: localID,
skipCheck: options.SkipCheck,
@ -90,7 +95,10 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy
intervalsStore: intervalsStore,
doRetrieve: options.DoRetrieve,
maxPeerServers: options.MaxPeerServers,
balanceMgr: balanceMgr,
}
streamer.setupSpec()
streamer.api = NewAPI(streamer)
delivery.getPeer = streamer.getPeer
streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, _ bool) (Server, error) {
@ -183,6 +191,14 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy
return streamer
}
func (r *Registry) setupSpec() {
r.createSpec()
r.createPriceOracle()
if r.balanceMgr != nil && !reflect.ValueOf(r.balanceMgr).IsNil() {
r.spec.Hook = protocols.NewAccounting(r.balanceMgr, r.prices)
}
}
// RegisterClient registers an incoming streamer constructor
func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, string, bool) (Client, error)) {
r.clientMu.Lock()
@ -448,7 +464,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)
@ -478,8 +494,13 @@ func (p *Peer) HandleMsg(ctx context.Context, msg interface{}) error {
case *WantedHashesMsg:
return p.handleWantedHashesMsg(ctx, msg)
case *ChunkDeliveryMsg:
return p.streamer.delivery.handleChunkDeliveryMsg(ctx, p, msg)
case *ChunkDeliveryMsgRetrieval:
//handling chunk delivery is the same for retrieval and syncing, so let's cast the msg
return p.streamer.delivery.handleChunkDeliveryMsg(ctx, p, ((*ChunkDeliveryMsg)(msg)))
case *ChunkDeliveryMsgSyncing:
//handling chunk delivery is the same for retrieval and syncing, so let's cast the msg
return p.streamer.delivery.handleChunkDeliveryMsg(ctx, p, ((*ChunkDeliveryMsg)(msg)))
case *RetrieveRequestMsg:
return p.streamer.delivery.handleRetrieveRequestMsg(ctx, p, msg)
@ -667,31 +688,92 @@ 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{},
ChunkDeliveryMsg{},
SubscribeErrorMsg{},
RequestSubscriptionMsg{},
QuitMsg{},
},
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: 7,
MaxMsgSize: 10 * 1024 * 1024,
Messages: []interface{}{
UnsubscribeMsg{},
OfferedHashesMsg{},
WantedHashesMsg{},
TakeoverProofMsg{},
SubscribeMsg{},
RetrieveRequestMsg{},
ChunkDeliveryMsgRetrieval{},
SubscribeErrorMsg{},
RequestSubscriptionMsg{},
QuitMsg{},
ChunkDeliveryMsgSyncing{},
},
}
r.spec = spec
}
//An accountable message needs some meta information attached to it
//in order to evaluate the correct price
type StreamerPrices struct {
priceMatrix map[uint64]*protocols.Price
registry *Registry
}
func (spo *StreamerPrices) Price(msg interface{}) *protocols.Price {
code, ok := spo.registry.spec.GetCode(msg)
if !ok {
//this could be handled more severely (panic) but let's be gentle?
return nil
}
price, ok := spo.priceMatrix[code]
if !ok {
return nil
}
return price
}
func (r *Registry) createPriceOracle() {
po := &StreamerPrices{
registry: r,
}
po.priceMatrix = make(map[uint64]*protocols.Price)
deliveryCode, ok := r.spec.GetCode(ChunkDeliveryMsgRetrieval{})
if !ok {
panic("Attempting to get message code for an expected message type, but code not found")
}
price := &protocols.Price{
Value: int64(-100),
PerByte: true,
}
po.priceMatrix[deliveryCode] = price
retrieveReqCode, ok := r.spec.GetCode(RetrieveRequestMsg{})
if !ok {
panic("Attempting to get message code for an expected message type, but code not found")
}
price = &protocols.Price{
Value: int64(10),
PerByte: false,
}
po.priceMatrix[retrieveReqCode] = price
r.prices = po
}
func (r *Registry) Protocols() []p2p.Protocol {
spec := r.spec
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)

8
swarm/network_test.go
View file

@ -40,9 +40,11 @@ import (
)
var (
loglevel = flag.Int("loglevel", 2, "verbosity of logs")
longrunning = flag.Bool("longrunning", false, "do run long-running tests")
waitKademlia = flag.Bool("waitkademlia", false, "wait for healthy kademlia before checking files availability")
loglevel = flag.Int("loglevel", 2, "verbosity of logs")
longrunning = flag.Bool("longrunning", false, "do run long-running tests")
waitKademlia = flag.Bool("waitkademlia", false, "wait for healthy kademlia before checking files availability")
printStats = flag.Bool("printstats", false, "print accounting stats to STDOUT")
bucketKeySwarm = simulation.BucketKey("swarm")
)
func init() {

90
swarm/swap/swap.go Normal file
View file

@ -0,0 +1,90 @@
// 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
}
//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
s.loadState(peer)
//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()]
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
}
//Get a peer's balance
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) {
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 {
s.stateStore.Get(peerID.String(), &peerBalance)
s.balances[peerID] = peerBalance
}
}
// New - swap constructor
func New(stateStore state.Store) (swap *Swap) {
swap = &Swap{
stateStore: stateStore,
balances: make(map[enode.ID]int64),
}
return
}

154
swarm/swap/swap_test.go Normal file
View file

@ -0,0 +1,154 @@
// Copyreeeeight 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"
"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()
log.PrintOrigins(true)
log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(colorable.NewColorableStderr(), log.TerminalFormat(true))))
}
//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,
//send a message (triggers to save to store),
//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)
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
}

429
swarm/swap_test.go Normal file
View file

@ -0,0 +1,429 @@
// 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 swarm
import (
"context"
"errors"
"fmt"
"io/ioutil"
"math/rand"
"os"
"strconv"
"sync"
"testing"
"time"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/api"
"github.com/ethereum/go-ethereum/swarm/log"
"github.com/ethereum/go-ethereum/swarm/network/simulation"
"github.com/ethereum/go-ethereum/swarm/storage"
)
//In TestSwapNetworkSymmetricFileUpload we set up a network with arbitrary number of nodes
//(16), and each of the nodes uploads a file of same size
//Afterwards we check that every node's balance WITH ANOTHER PEER
//has the same value but opposite sign
func TestSwapNetworkSymmetricFileUpload(t *testing.T) {
//default hardcoded network size
nodeCount := 16
//setup the simulation
//use a complete node setup via `NewSwam`
sim := simulation.New(map[string]simulation.ServiceFunc{
"swarm": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
config := api.NewConfig()
config.Port = strconv.Itoa(8500 + rand.Intn(9999))
dir, err := ioutil.TempDir("", "swap-network-test-node"+config.Port)
if err != nil {
return nil, nil, err
}
cleanup = func() {
err := os.RemoveAll(dir)
if err != nil {
log.Error("cleaning up swarm temp dir", "err", err)
}
}
config.Path = dir
privkey, err := crypto.GenerateKey()
if err != nil {
return nil, cleanup, err
}
config.Init(privkey)
//set Swap to be enabled for this test
config.SwapEnabled = true
swarm, err := NewSwarm(config, nil)
if err != nil {
return nil, cleanup, err
}
bucket.Store(bucketKeySwarm, swarm)
log.Info("new swarm", "bzzKey", config.BzzKey, "baseAddr", fmt.Sprintf("%x", swarm.bzz.BaseAddr()))
return swarm, cleanup, nil
},
})
defer sim.Close()
ctx := context.Background()
files := make([]file, 0)
var checkStatusM sync.Map
var nodeStatusM sync.Map
var totalFoundCount uint64
//connect all nodes in a chain
_, err := sim.AddNodesAndConnectChain(nodeCount)
if err != nil {
t.Fatal(err)
}
//run the simulation
result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
//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 {
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
}
log.Trace("file uploaded", "node", id, "key", key.String())
files = append(files, file{
addr: key,
data: data,
nodeID: id,
})
}
// 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 {
break
}
}
time.Sleep(5 * time.Second)
//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)
//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)
//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))
}
}
//NOTE: this are currently metrics over ALL nodes, not per node
fmt.Println(fmt.Sprintf("Total units credited: %d", metrics.Get("account.balance.credit").(metrics.Counter).Count()))
fmt.Println(fmt.Sprintf("Total units debited: %d", metrics.Get("account.balance.debit").(metrics.Counter).Count()))
fmt.Println(fmt.Sprintf("Total bytes credited: %d", metrics.Get("account.bytes.credit").(metrics.Counter).Count()))
fmt.Println(fmt.Sprintf("Total bytes debited: %d", metrics.Get("account.bytes.debit").(metrics.Counter).Count()))
fmt.Println(fmt.Sprintf("Cheques issued: %d", metrics.Get("account.cheques.issued").(metrics.Counter).Count()))
fmt.Println(fmt.Sprintf("Cheques received: %d", metrics.Get("account.cheques.received").(metrics.Counter).Count()))
fmt.Println(fmt.Sprintf("Number of messages credited: %d", metrics.Get("account.msg.credit").(metrics.Counter).Count()))
fmt.Println(fmt.Sprintf("Number of messages debited: %d", metrics.Get("account.msg.debit").(metrics.Counter).Count()))
fmt.Println(fmt.Sprintf("Peers dropped: %d", metrics.Get("account.peerdrops").(metrics.Counter).Count()))
fmt.Println(fmt.Sprintf("Number of times node dropped itself: %d", metrics.Get("account.selfdrops").(metrics.Counter).Count()))
}
//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
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 balanceKwithN+mapForSubK[k] != 0 && balanceKwithN != 0 {
log.Error(fmt.Sprintf("Expected balances to be a+b = 0 AND balance(a) != 0, but they are not, balance k with n: %d, balance n with k: %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)
}
log.Debug("test terminated")
}
//TestSwapNetworkAsymmetricFileUpload is a swap test too,
//but this time the number and size of files are random
func TestSwapNetworkAsymmetricFileUpload(t *testing.T) {
nodeCount := 16
sim := simulation.New(map[string]simulation.ServiceFunc{
"swarm": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
config := api.NewConfig()
config.Port = strconv.Itoa(8500 + rand.Intn(9999))
dir, err := ioutil.TempDir("", "swap-network-test-node"+config.Port)
if err != nil {
return nil, nil, err
}
cleanup = func() {
err := os.RemoveAll(dir)
if err != nil {
log.Error("cleaning up swarm temp dir", "err", err)
}
}
config.Path = dir
privkey, err := crypto.GenerateKey()
if err != nil {
return nil, cleanup, err
}
config.Init(privkey)
//enable swap
config.SwapEnabled = true
swarm, err := NewSwarm(config, nil)
if err != nil {
return nil, cleanup, err
}
bucket.Store(bucketKeySwarm, swarm)
log.Info("new swarm", "bzzKey", config.BzzKey, "baseAddr", fmt.Sprintf("%x", swarm.bzz.BaseAddr()))
return swarm, cleanup, nil
},
})
defer sim.Close()
ctx := context.Background()
files := make([]file, 0)
var checkStatusM sync.Map
var nodeStatusM sync.Map
var totalFoundCount uint64
_, err := sim.AddNodesAndConnectChain(nodeCount)
if err != nil {
t.Fatal(err)
}
//NOTE: maxFileSize is 4 kB, this in order to provide faster tests
//it would be interesting to run these tests with bigger files
//(to see how drop limits are affected etc.)
const maxFileSize = 1024 * 4 //1024 bytes * 4 = 4kB
const minfileSize = 1024
//pseudo random algo to define if a node will upload or not
//if a bit is 0, do not upload
pseudoRandomNum := rand.Int63()
pseudoRandomBitMask := strconv.FormatInt(pseudoRandomNum, 2)
result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
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]
})
for i, id := range nodeIDs {
//if the position in random num is 0, don't upload
if string(pseudoRandomBitMask[i]) != "0" {
size := rand.Intn(maxFileSize-minfileSize) + minfileSize
key, data, err := uploadRandomFileSize(sim.Service("swarm", id).(*Swarm), size)
if err != nil {
return err
}
log.Trace("file uploaded", "node", id, "key", key.String())
files = append(files, file{
addr: key,
data: data,
nodeID: id,
})
}
}
// 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 {
break
}
}
time.Sleep(5 * time.Second)
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 errors.New("no swarm")
}
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))
}
}
}
/*
Assuming that in this case, balances should be symmetric too I
*/
success := true
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 balanceKwithN+mapForSubK[k] != 0 && balanceKwithN != 0 {
log.Error(fmt.Sprintf("Expected balances to be a+b = 0 AND balance(a) != 0, but they are not, balance k with n: %d, balance n with k: %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)
}
log.Debug("test terminated")
}
// uploadFile, uploads a short file to the swarm instance
// using the api.Put method.
func uploadRandomFileSize(swarm *Swarm, size int) (storage.Address, string, error) {
b := make([]byte, size)
_, err := rand.Read(b)
if err != nil {
return nil, "", err
}
// uniqueness is very certain.
data := fmt.Sprintf("test content %s %x", time.Now().Round(0), b)
ctx := context.TODO()
k, wait, err := swarm.api.Put(ctx, data, "text/plain", false)
if err != nil {
return nil, "", err
}
if wait != nil {
err = wait(ctx)
}
return k, data, err
}

13
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,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
@ -181,7 +187,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)
@ -204,7 +210,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)
@ -341,7 +347,7 @@ func (self *Swarm) Start(srv *p2p.Server) error {
newaddr := self.bzz.UpdateLocalAddr([]byte(srv.Self().String()))
log.Info("Updated bzz local addr", "oaddr", fmt.Sprintf("%x", newaddr.OAddr), "uaddr", fmt.Sprintf("%s", newaddr.UAddr))
// set chequebook
if self.config.SwapEnabled {
if self.config.SwapEnabled && self.config.SwapAPI != "" {
ctx := context.Background() // The initial setup has no deadline.
err := self.SetChequebook(ctx)
if err != nil {
@ -439,6 +445,7 @@ func (self *Swarm) Protocols() (protos []p2p.Protocol) {
if self.ps != nil {
protos = append(protos, self.ps.Protocols()...)
}
return
}