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

swarm/swap: initial version

Browse source
This commit is contained in:
Fabio Barone 2018-08-09 12:34:36 -05:00
parent 317736d726
commit c89cc17453
7 changed files with 267 additions and 229 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -19,6 +19,8 @@ package stream
import (
"context"
"errors"
"math/big"
"time"
"fmt"
@ -28,6 +30,7 @@ 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"
)
@ -136,6 +139,11 @@ type RetrieveRequestMsg struct {
HopCount uint8
}
//TODO: what is the correct price
func (rrm *RetrieveRequestMsg) GetMsgPrice() (*big.Int, swap.EntryDirection) {
return big.NewInt(int64(4096)), swap.CreditEntry
}
func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *RetrieveRequestMsg) error {
log.Trace("received request", "peer", sp.ID(), "hash", req.Addr)
handleRetrieveRequestMsgCount.Inc(1)

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

@ -34,6 +34,8 @@ import (
"github.com/ethereum/go-ethereum/swarm/pot"
"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"
)
const (
@ -102,6 +104,12 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy
RegisterSwarmSyncerServer(streamer, syncChunkStore)
RegisterSwarmSyncerClient(streamer, syncChunkStore)
var err error
streamer.swap, err = swap.NewSwap(swap.NewDefaultSwapParams().Params)
if err != nil {
log.Error(err.Error())
}
if options.DoSync {
// latestIntC function ensures that
// - receiving from the in chan is not blocked by processing inside the for loop
@ -381,7 +389,7 @@ func (r *Registry) Run(p *network.BzzPeer) error {
}
}
return sp.Run(sp.HandleMsg)
return sp.Run(sp.HandleAccountedMsg)
}
// updateSyncing subscribes to SYNC streams by iterating over the
@ -456,8 +464,16 @@ func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error {
return r.Run(bp)
}
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:

6
swarm/swap/api.go
View file

@ -31,7 +31,7 @@ type APIMsg struct {
// Additional public methods accessible through API for pss
type API struct {
*Swap
*SwapProtocol
}
type SwapMetrics struct {
@ -40,8 +40,8 @@ type SwapMetrics struct {
type Cheque struct {
}
func NewAPI(swap *Swap) *API {
return &API{Swap: swap}
func NewAPI(swap *SwapProtocol) *API {
return &API{SwapProtocol: swap}
}
func (swapapi *API) Balance(ctx context.Context) (balance int, err error) {

274
swarm/swap/protocol.go
View file

@ -14,164 +14,186 @@
// 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/>.
// +build !nopssprotocol
package swap
import (
"context"
"fmt"
"sync"
"time"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/log"
)
const (
IsActiveProtocol = true
)
// Convenience wrapper for devp2p protocol messages for transport over pss
type ProtocolMsg struct {
Code uint64
Size uint32
Payload []byte
ReceivedAt time.Time
type SwapProtocol struct {
peersMu sync.RWMutex
peers map[discover.NodeID]*SwapPeer
}
// Creates a ProtocolMsg
func NewProtocolMsg(code uint64, msg interface{}) ([]byte, error) {
// Peer is the Peer extension for the streaming protocol
type SwapPeer struct {
*protocols.Peer
swapProtocol *SwapProtocol
}
rlpdata, err := rlp.EncodeToBytes(msg)
if err != nil {
return nil, err
func NewSwapProtocol() *SwapProtocol {
proto := &SwapProtocol{
peers: make(map[discover.NodeID]*SwapPeer),
}
return proto
}
// TODO verify that nested structs cannot be used in rlp
smsg := &ProtocolMsg{
Code: code,
Size: uint32(len(rlpdata)),
Payload: rlpdata,
// NewPeer is the constructor for Peer
func NewPeer(peer *protocols.Peer, swap *SwapProtocol) *SwapPeer {
p := &SwapPeer{
Peer: peer,
swapProtocol: swap,
}
return rlp.EncodeToBytes(smsg)
return p
}
// Protocol options to be passed to a new Protocol instance
//
// The parameters specify which encryption schemes to allow
type ProtocolParams struct {
type IssueChequeMsg struct {
}
// Convenience object for emulation devp2p over pss
type Protocol struct {
proto *p2p.Protocol
spec *protocols.Spec
RWPoolMu sync.Mutex
type RedeemChequeMsg struct {
}
/*
// Activates devp2p emulation over a specific pss topic
//
// One or both encryption schemes must be specified. If
// only one is specified, the protocol will not be valid
// for the other, and will make the message handler
// return errors
func RegisterProtocol(ps *Pss, topic *Topic, spec *protocols.Spec, targetprotocol *p2p.Protocol, options *ProtocolParams) (*Protocol, error) {
if !options.Asymmetric && !options.Symmetric {
return nil, fmt.Errorf("specify at least one of asymmetric or symmetric messaging mode")
}
pp := &Protocol{
Pss: ps,
proto: targetprotocol,
topic: topic,
spec: spec,
pubKeyRWPool: make(map[string]p2p.MsgReadWriter),
symKeyRWPool: make(map[string]p2p.MsgReadWriter),
Asymmetric: options.Asymmetric,
Symmetric: options.Symmetric,
}
return pp, nil
}
/////////////////////////////////////////////////////////////////////
// SECTION: node.Service interface
/////////////////////////////////////////////////////////////////////
*/
// Generic handler for incoming messages over devp2p emulation
//
// To be passed to pss.Register()
//
// Will run the protocol on a new incoming peer, provided that
// the encryption key of the message has a match in the internal
// pss keypool
//
// Fails if protocol is not valid for the message encryption scheme,
// if adding a new peer fails, or if the message is not a serialized
// p2p.Msg (which it always will be if it is sent from this object).
func (p *Protocol) Handle(msg []byte, peer *p2p.Peer, asymmetric bool, keyid string) error {
func (p *SwapProtocol) Start(srv *p2p.Server) error {
log.Debug("Started swap")
return nil
}
/*
// Runs an emulated pss Protocol on the specified peer,
// linked to a specific topic
// `key` and `asymmetric` specifies what encryption key
// to link the peer to.
// The key must exist in the pss store prior to adding the peer.
func (p *Protocol) AddPeer(peer *p2p.Peer, topic Topic, asymmetric bool, key string) (p2p.MsgReadWriter, error) {
rw := &PssReadWriter{
Pss: p.Pss,
rw: make(chan p2p.Msg),
spec: p.spec,
topic: p.topic,
key: key,
}
if asymmetric {
rw.sendFunc = p.Pss.SendAsym
} else {
rw.sendFunc = p.Pss.SendSym
}
if asymmetric {
p.Pss.pubKeyPoolMu.Lock()
if _, ok := p.Pss.pubKeyPool[key]; !ok {
return nil, fmt.Errorf("asym key does not exist: %s", key)
}
p.Pss.pubKeyPoolMu.Unlock()
p.RWPoolMu.Lock()
p.pubKeyRWPool[key] = rw
p.RWPoolMu.Unlock()
} else {
p.Pss.symKeyPoolMu.Lock()
if _, ok := p.Pss.symKeyPool[key]; !ok {
return nil, fmt.Errorf("symkey does not exist: %s", key)
}
p.Pss.symKeyPoolMu.Unlock()
p.RWPoolMu.Lock()
p.symKeyRWPool[key] = rw
p.RWPoolMu.Unlock()
}
go func() {
err := p.proto.Run(peer, rw)
log.Warn(fmt.Sprintf("pss vprotocol quit on %v topic %v: %v", peer, topic, err))
}()
return rw, nil
func (p *SwapProtocol) Stop() error {
log.Info("swap shutting down")
return nil
}
func (p *Protocol) RemovePeer(asymmetric bool, key string) {
log.Debug("closing pss peer", "asym", asymmetric, "key", key)
p.RWPoolMu.Lock()
defer p.RWPoolMu.Unlock()
if asymmetric {
rw := p.pubKeyRWPool[key].(*PssReadWriter)
rw.closed = true
delete(p.pubKeyRWPool, key)
} else {
rw := p.symKeyRWPool[key].(*PssReadWriter)
rw.closed = true
delete(p.symKeyRWPool, key)
var swapSpec = &protocols.Spec{
Name: swapProtocolName,
Version: swapVersion,
MaxMsgSize: defaultMaxMsgSize,
Messages: []interface{}{
SwapMsg{},
},
}
func (p *SwapProtocol) Protocols() []p2p.Protocol {
return []p2p.Protocol{
{
Name: swapSpec.Name,
Version: swapSpec.Version,
Length: swapSpec.Length(),
Run: p.Run,
},
}
}
// Uniform translation of protocol specifiers to topic
func ProtocolTopic(spec *protocols.Spec) Topic {
return BytesToTopic([]byte(fmt.Sprintf("%s:%d", spec.Name, spec.Version)))
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{
{
Namespace: "swap",
Version: "1.0",
Service: NewAPI(p),
Public: true,
},
}
return apis
}
//--------------------
func (swap *SwapProtocol) NodeInfo() interface{} {
return nil
}
func (swap *SwapProtocol) PeerInfo(id discover.NodeID) interface{} {
return nil
}
func (swap *SwapProtocol) Close() error {
return nil
}
func (swap *SwapProtocol) getPeer(peerId discover.NodeID) *SwapPeer {
swap.peersMu.RLock()
defer swap.peersMu.RUnlock()
return swap.peers[peerId]
}
func (swap *SwapProtocol) setPeer(peer *SwapPeer) {
swap.peersMu.Lock()
defer swap.peersMu.Unlock()
swap.peers[peer.ID()] = peer
metrics.GetOrRegisterGauge("registry.peers", nil).Update(int64(len(swap.peers)))
}
func (swap *SwapProtocol) deletePeer(peer *SwapPeer) {
swap.peersMu.Lock()
defer swap.peersMu.Unlock()
delete(swap.peers, peer.ID())
metrics.GetOrRegisterGauge("registry.peers", nil).Update(int64(len(swap.peers)))
swap.peersMu.Unlock()
}
func (swap *SwapProtocol) peersCount() (c int) {
swap.peersMu.Lock()
c = len(swap.peers)
swap.peersMu.Unlock()
return
}
func (p *SwapPeer) handleSwapMsg(ctx context.Context, msg interface{}) error {
switch msg := msg.(type) {
case *IssueChequeMsg:
return p.handleIssueChequeMsg(ctx, msg)
case *RedeemChequeMsg:
return p.handleRedeemChequeMsg(ctx, msg)
/*
case *QuitMsg:
return p.handleQuitMsg(msg)
*/
default:
return fmt.Errorf("unknown message type: %T", msg)
}
return nil
}
func (sp *SwapPeer) handleIssueChequeMsg(ctx context.Context, msg interface{}) (err error) {
return err
}
func (sp *SwapPeer) handleRedeemChequeMsg(ctx context.Context, msg interface{}) (err error) {
return err
}
*/

153
swarm/swap/swap.go
View file

@ -32,9 +32,7 @@ import (
"github.com/ethereum/go-ethereum/contracts/chequebook/contract"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/swarm/log"
whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
)
@ -53,8 +51,8 @@ var (
autoDepositBuffer = big.NewInt(100000000000000) // buffer that is surplus for fork protection etc (wei)
buyAt = big.NewInt(20000000000) // maximum chunk price host is willing to pay (wei)
sellAt = big.NewInt(20000000000) // minimum chunk price host requires (wei)
payAt = 100 // threshold that triggers payment {request} (units)
dropAt = 10000 // threshold that triggers disconnect (units)
payAt = big.NewInt(4096 * 10000) // threshold that triggers payment {request} (bytes)
dropAt = big.NewInt(4096 * 10000) // threshold that triggers disconnect (bytes)
)
const (
@ -66,10 +64,38 @@ const (
// Swift Automatic Payments
// a peer to peer micropayment system
type Swap struct {
lock sync.Mutex // mutex for balance access
balance int // units of chunk/retrieval request
lock sync.RWMutex
peers map[discover.NodeID]*swapPeer
local *Params // local peer's swap parameters
remote *Profile // remote peer's swap profile
}
type EntryDirection bool
const (
DebitEntry EntryDirection = true
CreditEntry EntryDirection = false
)
type SwapAccountedMsgType interface {
GetMsgPrice() (*big.Int, EntryDirection)
}
func (swap *Swap) AccountForMsg(ctx context.Context, msg interface{}, peer discover.NodeID) error {
if accounted, ok := msg.(SwapAccountedMsgType); ok {
if _, exists := swap.peers[peer]; !exists {
swap.lock.Lock()
swap.peers[peer] = &swapPeer{
peer: peer,
swapAccount: swap,
balance: big.NewInt(0),
}
swap.lock.Unlock()
}
price, direction := accounted.GetMsgPrice()
//TODO: Calculate total price and account
swap.peers[peer].AccountMsgForPeer(price, direction)
}
return nil
}
// Profile - public swap profile
@ -77,8 +103,8 @@ type Swap struct {
type Profile struct {
BuyAt *big.Int // accepted max price for chunk
SellAt *big.Int // offered sale price for chunk
PayAt uint // threshold that triggers payment request
DropAt uint // threshold that triggers disconnect
PayAt *big.Int // threshold that triggers payment request
DropAt *big.Int // threshold that triggers disconnect
}
// Strategy encapsulates parameters relating to
@ -130,15 +156,41 @@ type PayProfile struct {
lock sync.RWMutex
}
type swapPeer struct {
lock sync.RWMutex
peer discover.NodeID
swapAccount *Swap
balance *big.Int
}
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)
}
if sp.balance.Cmp(payAt) > -1 {
//TODO: Issue Cheque
}
if sp.balance.Cmp(dropAt) < 0 {
//TODO: Drop peer
}
log.Error(fmt.Sprintf("balance for peer %s: %s", sp.peer, sp.balance.String()))
}
// New - swap constructor
func NewSwap(local *Params) (swap *Swap, err error) {
swap = &Swap{
local: local,
peers: make(map[discover.NodeID]*swapPeer),
}
//swap.SetParams(local)
return
}
@ -150,8 +202,8 @@ func NewDefaultSwapParams() *LocalProfile {
Profile: &Profile{
BuyAt: buyAt,
SellAt: sellAt,
PayAt: uint(payAt),
DropAt: uint(dropAt),
PayAt: payAt,
DropAt: dropAt,
},
Strategy: &Strategy{
AutoCashInterval: autoCashInterval,
@ -179,70 +231,6 @@ func (lp *LocalProfile) Init(contract common.Address, prvkey *ecdsa.PrivateKey)
}
}
/////////////////////////////////////////////////////////////////////
// SECTION: node.Service interface
/////////////////////////////////////////////////////////////////////
func (p *Swap) Start(srv *p2p.Server) error {
log.Debug("Started swap")
return nil
}
func (p *Swap) Stop() error {
log.Info("swap shutting down")
return nil
}
var swapSpec = &protocols.Spec{
Name: swapProtocolName,
Version: swapVersion,
MaxMsgSize: defaultMaxMsgSize,
Messages: []interface{}{
SwapMsg{},
},
}
func (p *Swap) Protocols() []p2p.Protocol {
return []p2p.Protocol{
{
Name: swapSpec.Name,
Version: swapSpec.Version,
Length: swapSpec.Length(),
Run: p.Run,
},
}
}
func (p *Swap) Run(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
pp := protocols.NewPeer(peer, rw, swapSpec)
/*
p.fwdPoolMu.Lock()
p.fwdPool[peer.Info().ID] = pp
p.fwdPoolMu.Unlock()
*/
return pp.Run(p.handleSwapMsg)
}
func (p *Swap) APIs() []rpc.API {
apis := []rpc.API{
{
Namespace: "swap",
Version: "1.0",
Service: NewAPI(p),
Public: true,
},
}
return apis
}
// Filters incoming messages for processing or forwarding.
// Check if address partially matches
// If yes, it CAN be for us, and we process it
// Only passes error to pss protocol handler if payload is not valid pssmsg
func (p *Swap) handleSwapMsg(ctx context.Context, msg interface{}) error {
return nil
}
// Chequebook get's chequebook from the localProfile
func (lp *LocalProfile) Chequebook() *chequebook.Chequebook {
defer lp.lock.Unlock()
@ -351,12 +339,13 @@ func (lp *LocalProfile) newChequebookFromContract(path string, backend chequeboo
return nil
}
/*
// Add (n)
// n > 0 called when promised/provided n units of service
// n < 0 called when used/requested n units of service
func (swap *Swap) Add(n int) error {
defer swap.lock.Unlock()
swap.lock.Lock()
//defer swap.lock.Unlock()
//swap.lock.Lock()
swap.balance += n
if !swap.Sells && swap.balance > 0 {
log.Trace(fmt.Sprintf("<%v> remote peer cannot have debt (balance: %v)", swap.proto, swap.balance))
@ -379,11 +368,12 @@ func (swap *Swap) Add(n int) error {
// Balance accessor
func (swap *Swap) Balance() int {
defer swap.lock.Unlock()
swap.lock.Lock()
//defer swap.lock.Unlock()
//swap.lock.Lock()
return swap.balance
}
/*
// send (units) is called when payment is due
// In case of insolvency no promise is issued and sent, safe against fraud
// No return value: no error = payment is opportunistic = hang in till dropped
@ -431,3 +421,4 @@ func (swap *Swap) Receive(units int, promise Promise) error {
return nil
}
*/

9
swarm/swap/swap_test.go
View file

@ -71,7 +71,7 @@ func newServiceNode(port int, httpport int, wsport int, modules ...string) (*nod
return stack, nil
}
func TestBasicSwap(t *testing.T) {
func TestSwapProtocol(t *testing.T) {
// create the two nodes
stack_one, err := newServiceNode(p2pPort, 0, 0)
@ -83,14 +83,11 @@ func TestBasicSwap(t *testing.T) {
log.Crit("Create servicenode #2 fail", "err", err)
}
instance, err := NewSwap(NewDefaultSwapParams().Params)
if err != nil {
t.Fatal("Couldn't create Swap instance")
}
instance := NewSwapProtocol()
// wrapper function for servicenode to start the service
swapsvc := func(ctx *node.ServiceContext) (node.Service, error) {
return &API{
Swap: instance,
SwapProtocol: instance,
}, nil
}

4
swarm/swarm.go
View file

@ -209,10 +209,12 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e
self.bzz = network.NewBzz(bzzconfig, to, stateStore, stream.Spec, self.streamer.Run)
/*
self.swap, err = swap.NewSwap(config.Swap.Params)
if err != nil {
return nil, err
}
*/
// Pss = postal service over swarm (devp2p over bzz)
self.ps, err = pss.NewPss(to, config.Pss)
if err != nil {
@ -368,12 +370,14 @@ func (self *Swarm) Start(srv *p2p.Server) error {
}
log.Info("Swarm network started", "bzzaddr", fmt.Sprintf("%x", self.bzz.Hive.BaseAddr()))
/*
err = self.swap.Start(srv)
if err != nil {
log.Error("swap failed", "err", err)
return err
}
log.Debug("Swap accounting initialized")
*/
if self.ps != nil {
self.ps.Start(srv)