diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 0429c4dffc..29e6a2b938 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -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) diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 3861cfcf6a..7576e91da9 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -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: diff --git a/swarm/swap/api.go b/swarm/swap/api.go index 2de11d9efa..632a62433f 100644 --- a/swarm/swap/api.go +++ b/swarm/swap/api.go @@ -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) { diff --git a/swarm/swap/protocol.go b/swarm/swap/protocol.go index 63acbbe929..1744bc9ada 100644 --- a/swarm/swap/protocol.go +++ b/swarm/swap/protocol.go @@ -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 . -// +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 } -*/ diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go index 7b58331cbf..b67c53768e 100644 --- a/swarm/swap/swap.go +++ b/swarm/swap/swap.go @@ -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 - local *Params // local peer's swap parameters - remote *Profile // remote peer's swap profile + lock sync.RWMutex + peers map[discover.NodeID]*swapPeer + local *Params // local peer's swap parameters +} + +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 } +*/ diff --git a/swarm/swap/swap_test.go b/swarm/swap/swap_test.go index 289ba028be..2be0c5013f 100644 --- a/swarm/swap/swap_test.go +++ b/swarm/swap/swap_test.go @@ -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 } diff --git a/swarm/swarm.go b/swarm/swarm.go index e2585c6de3..7c6adb5e74 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -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 - } + /* + 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") + /* + 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)