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

swarm/swap: swap implemented as protocols/protocol.Peer extension

Browse source
This commit is contained in:
Fabio Barone 2018-08-15 13:57:03 -05:00
parent f8ec67a533
commit acaeb2d5ea
5 changed files with 113 additions and 88 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -30,7 +30,6 @@ import (
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/spancontext"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/swap"
opentracing "github.com/opentracing/opentracing-go"
)
@ -141,8 +140,8 @@ type RetrieveRequestMsg struct {
}
//TODO: what is the correct price
func (rrm *RetrieveRequestMsg) GetMsgPrice() (*big.Int, swap.EntryDirection) {
return big.NewInt(int64(4096)), swap.CreditEntry
func (rrm *RetrieveRequestMsg) GetMsgPrice() *big.Int {
return big.NewInt(int64(4096))
}
func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *RetrieveRequestMsg) error {

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

@ -30,6 +30,7 @@ import (
"github.com/ethereum/go-ethereum/swarm/spancontext"
"github.com/ethereum/go-ethereum/swarm/state"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/swap"
opentracing "github.com/opentracing/opentracing-go"
)
@ -50,7 +51,8 @@ func (e *notFoundError) Error() string {
// Peer is the Peer extension for the streaming protocol
type Peer struct {
*protocols.Peer
//*protocols.Peer
*swap.SwapPeer
streamer *Registry
pq *pq.PriorityQueue
serverMu sync.RWMutex
@ -72,7 +74,7 @@ type WrappedPriorityMsg struct {
// NewPeer is the constructor for Peer
func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer {
p := &Peer{
Peer: peer,
SwapPeer: swap.NewSwapPeer(peer, streamer.swap),
pq: pq.New(int(PriorityQueue), PriorityQueueCap),
streamer: streamer,
servers: make(map[Stream]*server),

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

@ -394,7 +394,7 @@ func (r *Registry) Run(p *network.BzzPeer) error {
}
}
return sp.Run(sp.HandleAccountedMsg)
return sp.RunAccountedProtocol(sp.HandleMsg)
}
// updateSyncing subscribes to SYNC streams by iterating over the
@ -471,16 +471,8 @@ func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error {
return r.Run(bzzPeer)
}
func (p *Peer) HandleAccountedMsg(ctx context.Context, msg interface{}) error {
err := p.handleMsg(ctx, msg)
if _, ok := msg.(swap.SwapAccountedMsgType); ok && err == nil {
p.streamer.swap.AccountForMsg(ctx, msg, p.ID())
}
return err
}
// HandleMsg is the message handler that delegates incoming messages
func (p *Peer) handleMsg(ctx context.Context, msg interface{}) error {
func (p *Peer) HandleMsg(ctx context.Context, msg interface{}) error {
switch msg := msg.(type) {
case *SubscribeMsg:

54
swarm/swap/protocol.go
View file

@ -35,25 +35,25 @@ const (
type SwapProtocol struct {
peersMu sync.RWMutex
peers map[discover.NodeID]*SwapPeer
peers map[discover.NodeID]*SwapProtocolPeer
}
// Peer is the Peer extension for the streaming protocol
type SwapPeer struct {
type SwapProtocolPeer struct {
*protocols.Peer
swapProtocol *SwapProtocol
}
func NewSwapProtocol() *SwapProtocol {
proto := &SwapProtocol{
peers: make(map[discover.NodeID]*SwapPeer),
peers: make(map[discover.NodeID]*SwapProtocolPeer),
}
return proto
}
// NewPeer is the constructor for Peer
func NewPeer(peer *protocols.Peer, swap *SwapProtocol) *SwapPeer {
p := &SwapPeer{
func NewPeer(peer *protocols.Peer, swap *SwapProtocol) *SwapProtocolPeer {
p := &SwapProtocolPeer{
Peer: peer,
swapProtocol: swap,
}
@ -100,19 +100,6 @@ func (p *SwapProtocol) Protocols() []p2p.Protocol {
}
}
func (swap *SwapProtocol) DebitByteCount(peer *SwapPeer, numberOfBytes int) error {
return nil
}
func (swap *SwapProtocol) Run(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
p := protocols.NewPeer(peer, rw, swapSpec)
sp := NewPeer(p, swap)
swap.setPeer(sp)
defer swap.deletePeer(sp)
defer swap.Close()
return sp.Run(sp.handleSwapMsg)
}
func (p *SwapProtocol) APIs() []rpc.API {
apis := []rpc.API{
{
@ -125,7 +112,17 @@ func (p *SwapProtocol) APIs() []rpc.API {
return apis
}
//--------------------
/////////////////////////////////////////////////////////////////////
// SECTION: p2p.protocol interface
/////////////////////////////////////////////////////////////////////
func (swap *SwapProtocol) Run(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
p := protocols.NewPeer(peer, rw, swapSpec)
sp := NewPeer(p, swap)
swap.setPeer(sp)
defer swap.deletePeer(sp)
defer swap.Close()
return sp.Run(sp.handleSwapMsg)
}
func (swap *SwapProtocol) NodeInfo() interface{} {
return nil
@ -135,18 +132,23 @@ func (swap *SwapProtocol) PeerInfo(id discover.NodeID) interface{} {
return nil
}
func (swap *SwapProtocol) Close() error {
//------------------------------------------------------------------------------------------
func (swap *SwapProtocol) DebitByteCount(peer *SwapProtocolPeer, numberOfBytes int) error {
return nil
}
func (swap *SwapProtocol) getPeer(peerId discover.NodeID) *SwapPeer {
func (swap *SwapProtocol) Close() {
}
func (swap *SwapProtocol) getPeer(peerId discover.NodeID) *SwapProtocolPeer {
swap.peersMu.RLock()
defer swap.peersMu.RUnlock()
return swap.peers[peerId]
}
func (swap *SwapProtocol) setPeer(peer *SwapPeer) {
func (swap *SwapProtocol) setPeer(peer *SwapProtocolPeer) {
swap.peersMu.Lock()
defer swap.peersMu.Unlock()
@ -154,7 +156,7 @@ func (swap *SwapProtocol) setPeer(peer *SwapPeer) {
metrics.GetOrRegisterGauge("registry.peers", nil).Update(int64(len(swap.peers)))
}
func (swap *SwapProtocol) deletePeer(peer *SwapPeer) {
func (swap *SwapProtocol) deletePeer(peer *SwapProtocolPeer) {
swap.peersMu.Lock()
defer swap.peersMu.Unlock()
@ -170,7 +172,7 @@ func (swap *SwapProtocol) peersCount() (c int) {
return
}
func (p *SwapPeer) handleSwapMsg(ctx context.Context, msg interface{}) error {
func (p *SwapProtocolPeer) handleSwapMsg(ctx context.Context, msg interface{}) error {
switch msg := msg.(type) {
case *IssueChequeMsg:
@ -190,10 +192,10 @@ func (p *SwapPeer) handleSwapMsg(ctx context.Context, msg interface{}) error {
return nil
}
func (sp *SwapPeer) handleIssueChequeMsg(ctx context.Context, msg interface{}) (err error) {
func (sp *SwapProtocolPeer) handleIssueChequeMsg(ctx context.Context, msg interface{}) (err error) {
return err
}
func (sp *SwapPeer) handleRedeemChequeMsg(ctx context.Context, msg interface{}) (err error) {
func (sp *SwapProtocolPeer) handleRedeemChequeMsg(ctx context.Context, msg interface{}) (err error) {
return err
}

124
swarm/swap/swap.go
View file

@ -33,6 +33,7 @@ import (
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/swarm/log"
"github.com/ethereum/go-ethereum/swarm/state"
whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
@ -67,10 +68,19 @@ const (
type Swap struct {
stateStore state.Store
lock sync.RWMutex
peers map[discover.NodeID]*swapPeer
peers map[discover.NodeID]*SwapPeer
local *Params // local peer's swap parameters
}
type SwapPeer struct {
*protocols.Peer
lock sync.RWMutex
swapAccount *Swap
handlerFunc func(context.Context, interface{}) error
balance *big.Int
storeID string
}
type EntryDirection bool
const (
@ -79,26 +89,20 @@ const (
)
type SwapAccountedMsgType interface {
GetMsgPrice() (*big.Int, EntryDirection)
GetMsgPrice() *big.Int
}
func (swap *Swap) AccountForMsg(ctx context.Context, msg interface{}, peer discover.NodeID) error {
func (sp *SwapPeer) RunAccountedProtocol(protocolHandler func(ctx context.Context, msg interface{}) error) error {
sp.handlerFunc = protocolHandler
return sp.Run(sp.handleAccountedMsg)
}
func (sp *SwapPeer) doAccountMsg(ctx context.Context, msg interface{}, direction EntryDirection) error {
if accounted, ok := msg.(SwapAccountedMsgType); ok {
if _, exists := swap.peers[peer]; !exists {
balance := big.NewInt(0)
swap.stateStore.Get(peer.String()[:24]+"-swap", &balance)
swap.lock.Lock()
swap.peers[peer] = &swapPeer{
peer: peer,
swapAccount: swap,
balance: balance,
storeID: peer.String()[:24] + "-swap",
}
swap.lock.Unlock()
}
price, direction := accounted.GetMsgPrice()
price := accounted.GetMsgPrice()
//TODO: Calculate total price and account
swap.peers[peer].AccountMsgForPeer(price, direction)
sp.AccountMsgForPeer(price, direction)
}
return nil
}
@ -110,6 +114,61 @@ func (swap *Swap) GetPeerBalance(peer discover.NodeID) *big.Int {
return nil
}
func (sp *SwapPeer) handleAccountedMsg(ctx context.Context, msg interface{}) error {
err := sp.handlerFunc(ctx, msg)
if _, ok := msg.(SwapAccountedMsgType); ok && err == nil {
sp.doAccountMsg(ctx, msg, CreditEntry)
}
return err
}
func (sp *SwapPeer) Send(ctx context.Context, msg interface{}) error {
err := sp.Peer.Send(ctx, msg)
if _, ok := msg.(SwapAccountedMsgType); ok && err == nil {
sp.doAccountMsg(ctx, msg, DebitEntry)
}
return err
}
//The balance is accounted from the point of view of the local node
//Thus, we credit the balance and increase it when the amount is in favor of the local node
//We debit the balance and decrease it when the amount is in favor of the remote peer
func (sp *SwapPeer) AccountMsgForPeer(price *big.Int, direction EntryDirection) {
sp.lock.Lock()
defer sp.lock.Unlock()
//local node is being credited (in its favor), so its balance increases
if direction == CreditEntry {
sp.balance = sp.balance.Add(sp.balance, price)
//local node is being debited (in favor of remote peer), so its balance decreases
} else if direction == DebitEntry {
sp.balance = sp.balance.Sub(sp.balance, price)
}
//TODO: save to store here? init store?
sp.swapAccount.stateStore.Put(sp.storeID, sp.balance)
if sp.balance.Cmp(payAt) > -1 {
//TODO: Issue Cheque
}
if sp.balance.Cmp(dropAt) < 0 {
//TODO: Drop peer
}
log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.ID(), sp.balance.String()))
}
func NewSwapPeer(peer *protocols.Peer, swap *Swap) *SwapPeer {
balance := big.NewInt(0)
swap.stateStore.Get(peer.String()[:24]+"-swap", &balance)
sp := &SwapPeer{
Peer: peer,
swapAccount: swap,
balance: balance,
storeID: peer.String()[:24] + "-swap",
}
swap.lock.Lock()
defer swap.lock.Unlock()
swap.peers[peer.ID()] = sp
return sp
}
// Profile - public swap profile
// public parameters for SWAP, serializable config struct passed in handshake
type Profile struct {
@ -168,42 +227,13 @@ type PayProfile struct {
lock sync.RWMutex
}
type swapPeer struct {
lock sync.RWMutex
peer discover.NodeID
swapAccount *Swap
balance *big.Int
storeID string
}
func (sp *swapPeer) AccountMsgForPeer(price *big.Int, direction EntryDirection) {
sp.lock.Lock()
defer sp.lock.Unlock()
//the peer is being credited (in its favor), so its balance increases
if direction == CreditEntry {
sp.balance = sp.balance.Add(sp.balance, price)
//the peer is being debited (in local favor), so its balance decreases
} else if direction == DebitEntry {
sp.balance = sp.balance.Sub(sp.balance, price)
}
//TODO: save to store here? init store?
sp.swapAccount.stateStore.Put(sp.storeID, sp.balance)
if sp.balance.Cmp(payAt) > -1 {
//TODO: Issue Cheque
}
if sp.balance.Cmp(dropAt) < 0 {
//TODO: Drop peer
}
log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.peer, sp.balance.String()))
}
// New - swap constructor
func NewSwap(local *Params, stateStore state.Store) (swap *Swap, err error) {
swap = &Swap{
local: local,
stateStore: stateStore,
peers: make(map[discover.NodeID]*swapPeer),
peers: make(map[discover.NodeID]*SwapPeer),
}
//swap.SetParams(local)