swarm/swap: rebased on latest master

This commit is contained in:
Fabio Barone 2018-10-02 16:34:57 -05:00
parent 12d825e379
commit 3d234cdbbb
7 changed files with 55 additions and 64 deletions

View file

@ -21,7 +21,7 @@ import (
"errors" "errors"
"math/big" "math/big"
"github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/enode"
) )
var ( var (
@ -44,7 +44,7 @@ func NewAPI(swap *Swap) *API {
} }
//Get the balance for this node with a specific peer //Get the balance for this node with a specific peer
func (swapapi *API) BalanceWithPeer(ctx context.Context, peer discover.NodeID) (balance *big.Int, err error) { func (swapapi *API) BalanceWithPeer(ctx context.Context, peer enode.ID) (balance *big.Int, err error) {
balance = swapapi.swap.peers[peer] balance = swapapi.swap.peers[peer]
if balance == nil { if balance == nil {
err = ErrNoSuchPeerAccounting err = ErrNoSuchPeerAccounting

View file

@ -19,15 +19,15 @@ package swap
import ( import (
"math/big" "math/big"
"github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/swarm/state" "github.com/ethereum/go-ethereum/swarm/state"
) )
type ChequeManager struct { type ChequeManager struct {
stateStore state.Store stateStore state.Store
serialPerNode map[discover.NodeID]uint64 serialPerNode map[enode.ID]uint64
openDebitCheques map[discover.NodeID][]*Cheque openDebitCheques map[enode.ID][]*Cheque
openCreditCheques map[discover.NodeID][]*Cheque openCreditCheques map[enode.ID][]*Cheque
} }
type Cheque struct { type Cheque struct {
@ -35,20 +35,20 @@ type Cheque struct {
timeout uint64 timeout uint64
amount *big.Int amount *big.Int
sumCumulated *big.Int sumCumulated *big.Int
beneficiary discover.NodeID //this should probably be common.Address? beneficiary enode.ID //this should probably be common.Address?
} }
func NewChequeManager(stateStore state.Store) *ChequeManager { func NewChequeManager(stateStore state.Store) *ChequeManager {
return &ChequeManager{ return &ChequeManager{
stateStore: stateStore, stateStore: stateStore,
//TODO: restore from state store //TODO: restore from state store
serialPerNode: make(map[discover.NodeID]uint64), serialPerNode: make(map[enode.ID]uint64),
openDebitCheques: make(map[discover.NodeID][]*Cheque), openDebitCheques: make(map[enode.ID][]*Cheque),
openCreditCheques: make(map[discover.NodeID][]*Cheque), openCreditCheques: make(map[enode.ID][]*Cheque),
} }
} }
func (mgr *ChequeManager) CreateCheque(beneficiary discover.NodeID, amount *big.Int) *Cheque { func (mgr *ChequeManager) CreateCheque(beneficiary enode.ID, amount *big.Int) *Cheque {
mgr.serialPerNode[beneficiary]++ mgr.serialPerNode[beneficiary]++
cheque := &Cheque{ cheque := &Cheque{
serial: mgr.serialPerNode[beneficiary], serial: mgr.serialPerNode[beneficiary],

View file

@ -22,7 +22,7 @@ import (
"sync" "sync"
"github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/log" "github.com/ethereum/go-ethereum/swarm/log"
@ -40,7 +40,7 @@ const (
//For a better understanding, please read the Swap network paper "Generalized Swap swear and swindle games" //For a better understanding, please read the Swap network paper "Generalized Swap swear and swindle games"
type Protocol struct { type Protocol struct {
peersMu sync.RWMutex peersMu sync.RWMutex
peers map[discover.NodeID]*Peer peers map[enode.ID]*Peer
} }
//This is the peer representing a participant in this protocol //This is the peer representing a participant in this protocol
@ -51,7 +51,7 @@ type Peer struct {
//Create a new protocol instance //Create a new protocol instance
func NewProtocol() *Protocol { func NewProtocol() *Protocol {
proto := &Protocol{ proto := &Protocol{
peers: make(map[discover.NodeID]*Peer), peers: make(map[enode.ID]*Peer),
} }
return proto return proto
} }
@ -140,7 +140,7 @@ func (s *Swap) NodeInfo() interface{} {
return nil return nil
} }
func (s *Swap) PeerInfo(id discover.NodeID) interface{} { func (s *Swap) PeerInfo(id enode.ID) interface{} {
return nil return nil
} }
@ -148,7 +148,7 @@ func (s *Swap) PeerInfo(id discover.NodeID) interface{} {
func (proto *Protocol) Close() { func (proto *Protocol) Close() {
} }
func (proto *Protocol) getPeer(peerId discover.NodeID) *Peer { func (proto *Protocol) getPeer(peerId enode.ID) *Peer {
proto.peersMu.RLock() proto.peersMu.RLock()
defer proto.peersMu.RUnlock() defer proto.peersMu.RUnlock()

View file

@ -22,10 +22,9 @@ import (
"fmt" "fmt"
"math/big" "math/big"
"sync" "sync"
"time"
"github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/swarm/log" "github.com/ethereum/go-ethereum/swarm/log"
"github.com/ethereum/go-ethereum/swarm/network/stream" "github.com/ethereum/go-ethereum/swarm/network/stream"
"github.com/ethereum/go-ethereum/swarm/state" "github.com/ethereum/go-ethereum/swarm/state"
@ -45,23 +44,17 @@ var (
ErrInsufficientFunds = errors.New("Insufficient funds") ErrInsufficientFunds = errors.New("Insufficient funds")
) )
const (
chequebookDeployRetries = 5
chequebookDeployDelay = 1 * time.Second // delay between retries
)
// SwAP Swarm Accounting Protocol // SwAP Swarm Accounting Protocol
// a peer to peer micropayment system // a peer to peer micropayment system
// A node maintains an individual balance with every peer // A node maintains an individual balance with every peer
// Only messages which have a price will be accounted for // Only messages which have a price will be accounted for
type Swap struct { type Swap struct {
priceOracle PriceOracle priceOracle PriceOracle //the price oracle facilitates message pricing to the swap instance
chequeManager *ChequeManager chequeManager *ChequeManager //cheque manager keeps track of issued cheques
stateStore state.Store stateStore state.Store //stateStore is needed in order to keep balances across sessions
lock sync.RWMutex lock sync.RWMutex //lock the balances
peers map[discover.NodeID]*big.Int peers map[enode.ID]*big.Int //map of balances for each peer
p2pServer *p2p.Server protocol *Protocol //reference to the cheque exchange protocol
protocol *Protocol
} }
//PriceOracle is responsible to maintain the price matrix for accounted messages, //PriceOracle is responsible to maintain the price matrix for accounted messages,
@ -197,37 +190,27 @@ func (s *Swap) handleMsgEvent(event *p2p.PeerEvent) {
if !s.priceOracle.IsAccountedMsg(event) { if !s.priceOracle.IsAccountedMsg(event) {
return return
} }
s.AccountMsgForPeer(event) s.accountMsgForPeer(event)
} }
//Do the accounting //Do the accounting
//Depending on the charging type of the message (set in the PriceTag), //Depending on the charging type of the message (set in the PriceTag),
//it will charge the sender or receiver //it will charge the sender or receiver
func (s *Swap) AccountMsgForPeer(event *p2p.PeerEvent) { func (s *Swap) accountMsgForPeer(event *p2p.PeerEvent) {
price, direction := s.priceOracle.GetPriceForMsg(event) price, direction := s.priceOracle.GetPriceForMsg(event)
if price == nil { if price == nil {
//TODO what to do in this case? Should not happen //TODO what to do in this case? Should not happen
log.Crit("Price is nil; this should have been accounted for but somehow it failed") panic("Price is nil; this should have been accounted for but somehow it failed")
} }
if direction == ChargeSender { if chargeLocal := (direction == ChargeSender) == (event.Type == p2p.PeerEventTypeMsgSend); chargeLocal {
//we are sending a ChargeSender message, thus debit us and credit remote //debit local node and credit remote
if event.Type == p2p.PeerEventTypeMsgSend {
s.chargeLocal(event, price) s.chargeLocal(event, price)
//we are receiving a ChargeSender message, so credit us and debit remote
} else if event.Type == p2p.PeerEventTypeMsgRecv {
s.chargeRemote(event, price)
}
} else { } else {
//we are receiving a ChargeReceiver message, thus debit us and credit remote //credit local node and debit remote
if event.Type == p2p.PeerEventTypeMsgRecv {
s.chargeLocal(event, price)
//we are sending a ChargeReceiver message, thus credit us and debit remote
} else if event.Type == p2p.PeerEventTypeMsgSend {
s.chargeRemote(event, price) s.chargeRemote(event, price)
} }
} }
}
//Debit us and credit remote //Debit us and credit remote
func (s *Swap) chargeLocal(event *p2p.PeerEvent, amount *big.Int) { func (s *Swap) chargeLocal(event *p2p.PeerEvent, amount *big.Int) {
@ -304,7 +287,7 @@ func (s *Swap) chargeRemote(event *p2p.PeerEvent, amount *big.Int) {
} }
//get a peer's balance //get a peer's balance
func (swap *Swap) GetPeerBalance(peer discover.NodeID) *big.Int { func (swap *Swap) GetPeerBalance(peer enode.ID) *big.Int {
swap.lock.RLock() swap.lock.RLock()
defer swap.lock.RUnlock() defer swap.lock.RUnlock()
if p, ok := swap.peers[peer]; ok { if p, ok := swap.peers[peer]; ok {
@ -315,7 +298,7 @@ func (swap *Swap) GetPeerBalance(peer discover.NodeID) *big.Int {
//Issue a cheque for the remote peer. Happens if we are indebted with the peer //Issue a cheque for the remote peer. Happens if we are indebted with the peer
//and crossed the payment threshold //and crossed the payment threshold
func (s *Swap) issueCheque(ctx context.Context, id discover.NodeID) error { func (s *Swap) issueCheque(ctx context.Context, id enode.ID) error {
amount := &big.Int{} amount := &big.Int{}
cheque := s.chequeManager.CreateCheque(id, amount.Abs(payAt)) cheque := s.chequeManager.CreateCheque(id, amount.Abs(payAt))
msg := IssueChequeMsg{ msg := IssueChequeMsg{
@ -337,7 +320,7 @@ func New(stateStore state.Store) (swap *Swap) {
swap = &Swap{ swap = &Swap{
chequeManager: NewChequeManager(stateStore), chequeManager: NewChequeManager(stateStore),
stateStore: stateStore, stateStore: stateStore,
peers: make(map[discover.NodeID]*big.Int), peers: make(map[enode.ID]*big.Int),
priceOracle: priceOracle, priceOracle: priceOracle,
protocol: NewProtocol(), protocol: NewProtocol(),
} }

View file

@ -24,7 +24,6 @@ import (
"math/big" "math/big"
"os" "os"
"path/filepath" "path/filepath"
"sync"
"testing" "testing"
"time" "time"
@ -42,7 +41,6 @@ var (
p2pPort = 30100 p2pPort = 30100
ipcpath = ".swarm.ipc" ipcpath = ".swarm.ipc"
datadirPrefix = ".data_" datadirPrefix = ".data_"
stackW = &sync.WaitGroup{}
loglevel = flag.Int("loglevel", 2, "verbosity of logs") loglevel = flag.Int("loglevel", 2, "verbosity of logs")
) )

View file

@ -1,3 +1,19 @@
// 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 package swarm
import ( import (
@ -14,7 +30,7 @@ import (
"github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/api" "github.com/ethereum/go-ethereum/swarm/api"
"github.com/ethereum/go-ethereum/swarm/log" "github.com/ethereum/go-ethereum/swarm/log"
@ -121,7 +137,7 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) {
//every node has a map to all nodes it had interactions //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 //each entry in the map is a map of the other node with all the balances
balancesMap := make(map[discover.NodeID]map[discover.NodeID]*big.Int) balancesMap := make(map[enode.ID]map[enode.ID]*big.Int)
//iterate all nodes //iterate all nodes
for _, node := range sim.NodeIDs() { for _, node := range sim.NodeIDs() {
@ -133,7 +149,7 @@ func TestSwapNetworkSymmetricFileUpload(t *testing.T) {
swarm := item.(*Swarm) swarm := item.(*Swarm)
//submap for each node is a map of all nodes with the balance for that node //submap for each node is a map of all nodes with the balance for that node
subBalances := make(map[discover.NodeID]*big.Int) subBalances := make(map[enode.ID]*big.Int)
//iterate all nodes again... //iterate all nodes again...
//get all balances with other peers for every node //get all balances with other peers for every node
@ -295,7 +311,7 @@ func TestSwapNetworkAsymmetricFileUpload(t *testing.T) {
} }
}) })
balancesMap := make(map[discover.NodeID]map[discover.NodeID]*big.Int) balancesMap := make(map[enode.ID]map[enode.ID]*big.Int)
for _, node := range sim.NodeIDs() { for _, node := range sim.NodeIDs() {
item, ok := sim.NodeItem(node, bucketKeySwarm) item, ok := sim.NodeItem(node, bucketKeySwarm)
@ -305,7 +321,7 @@ func TestSwapNetworkAsymmetricFileUpload(t *testing.T) {
} }
swarm := item.(*Swarm) swarm := item.(*Swarm)
subBalances := make(map[discover.NodeID]*big.Int) subBalances := make(map[enode.ID]*big.Int)
for _, n := range sim.NodeIDs() { for _, n := range sim.NodeIDs() {
if node == n { if node == n {

View file

@ -178,12 +178,6 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e
if err := nodeID.UnmarshalText([]byte(config.NodeID)); err != nil { if err := nodeID.UnmarshalText([]byte(config.NodeID)); err != nil {
return nil, err return nil, err
} }
if config.SwapEnabled {
self.swap, err = swap.New(stateStore)
if err != nil {
return nil, err
}
}
self.streamer = stream.NewRegistry(nodeID, delivery, self.netStore, stateStore, &stream.RegistryOptions{ self.streamer = stream.NewRegistry(nodeID, delivery, self.netStore, stateStore, &stream.RegistryOptions{
SkipCheck: config.DeliverySkipCheck, SkipCheck: config.DeliverySkipCheck,
DoSync: config.SyncEnabled, DoSync: config.SyncEnabled,