From acaeb2d5ea7965fc6dcb0a0e5b04d106dacfd32e Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Wed, 15 Aug 2018 13:57:03 -0500 Subject: [PATCH] swarm/swap: swap implemented as protocols/protocol.Peer extension --- swarm/network/stream/delivery.go | 5 +- swarm/network/stream/peer.go | 6 +- swarm/network/stream/stream.go | 12 +-- swarm/swap/protocol.go | 54 +++++++------- swarm/swap/swap.go | 124 +++++++++++++++++++------------ 5 files changed, 113 insertions(+), 88 deletions(-) diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index fde8ed0477..4ed7a25c69 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -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 { diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index 80b9ab711a..ce861b67d4 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -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), diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 70cb238974..9d559eeefd 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -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: diff --git a/swarm/swap/protocol.go b/swarm/swap/protocol.go index 1744bc9ada..b0a27983ec 100644 --- a/swarm/swap/protocol.go +++ b/swarm/swap/protocol.go @@ -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 } diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go index 62b8b2bdd0..36b2948677 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -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)