swarm/swap: initial version

swarm/network/stream/delivery.go

@@ -19,6 +19,7 @@ package stream
 import (
 	"context"
 	"errors"
+	"math/big"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -29,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"
 )
 
@@ -138,6 +140,11 @@ type RetrieveRequestMsg struct {
 	SkipCheck bool
 }
 
+//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)

swarm/network/stream/stream.go

@@ -35,6 +35,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,6 +51,7 @@ const (
 
 // Registry registry for outgoing and incoming streamer constructors
 type Registry struct {
+	swap           *swap.Swap
 	api            *API
 	addr           *network.BzzAddr
 	skipCheck      bool
@@ -101,6 +103,12 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i
 	RegisterSwarmSyncerServer(streamer, db)
 	RegisterSwarmSyncerClient(streamer, db)
 
+	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
@@ -390,7 +398,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
@@ -467,8 +475,16 @@ 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:

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 {
+func NewAPI(swap *SwapProtocol) *API {
-	return &API{Swap: swap}
+	return &API{SwapProtocol: swap}
 }
 
 func (swapapi *API) Balance(ctx context.Context) (balance int, err error) {

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 <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 SwapProtocol struct {
-type ProtocolMsg struct {
+	peersMu sync.RWMutex
-	Code       uint64
+	peers   map[discover.NodeID]*SwapPeer
-	Size       uint32
-	Payload    []byte
-	ReceivedAt time.Time
 }
 
-// Creates a ProtocolMsg
+// Peer is the Peer extension for the streaming protocol
-func NewProtocolMsg(code uint64, msg interface{}) ([]byte, error) {
+type SwapPeer struct {
-
+	*protocols.Peer
-	rlpdata, err := rlp.EncodeToBytes(msg)
+	swapProtocol *SwapProtocol
-	if err != nil {
-		return nil, err
 }
 
-	// TODO verify that nested structs cannot be used in rlp
+func NewSwapProtocol() *SwapProtocol {
-	smsg := &ProtocolMsg{
+	proto := &SwapProtocol{
-		Code:    code,
+		peers: make(map[discover.NodeID]*SwapPeer),
-		Size:    uint32(len(rlpdata)),
+	}
-		Payload: rlpdata,
+	return proto
 }
 
-	return rlp.EncodeToBytes(smsg)
+// NewPeer is the constructor for Peer
+func NewPeer(peer *protocols.Peer, swap *SwapProtocol) *SwapPeer {
+	p := &SwapPeer{
+		Peer:         peer,
+		swapProtocol: swap,
+	}
+	return p
 }
 
-// Protocol options to be passed to a new Protocol instance
+type IssueChequeMsg struct {
-//
-// The parameters specify which encryption schemes to allow
-type ProtocolParams struct {
 }
 
-// Convenience object for emulation devp2p over pss
+type RedeemChequeMsg struct {
-type Protocol struct {
-	proto    *p2p.Protocol
-	spec     *protocols.Spec
-	RWPoolMu sync.Mutex
 }
 
-/*
+/////////////////////////////////////////////////////////////////////
-// Activates devp2p emulation over a specific pss topic
+// SECTION: node.Service interface
-//
+/////////////////////////////////////////////////////////////////////
-// 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
-}
 
-*/
+func (p *SwapProtocol) Start(srv *p2p.Server) error {
-// Generic handler for incoming messages over devp2p emulation
+	log.Debug("Started swap")
-//
-// 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 {
 	return nil
 }
 
+func (p *SwapProtocol) Stop() error {
+	log.Info("swap shutting down")
+	return nil
+}
+
+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,
+		},
+	}
+}
+
+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)
+
 	/*
-// Runs an emulated pss Protocol on the specified peer,
+		case *QuitMsg:
-// linked to a specific topic
+			return p.handleQuitMsg(msg)
-// `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 *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)
-	}
-}
-
-// Uniform translation of protocol specifiers to topic
-func ProtocolTopic(spec *protocols.Spec) Topic {
-	return BytesToTopic([]byte(fmt.Sprintf("%s:%d", spec.Name, spec.Version)))
-}
 	*/
+
+	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
+}

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/discover"
-	"github.com/ethereum/go-ethereum/p2p/protocols"
-	"github.com/ethereum/go-ethereum/rpc"
 	"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)
+	payAt                = big.NewInt(4096 * 10000)    // threshold that triggers payment {request} (bytes)
-	dropAt               = 10000                       // threshold that triggers disconnect (units)
+	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
+	lock  sync.RWMutex
-	balance int        // units of chunk/retrieval request
+	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
+	PayAt  *big.Int // threshold that triggers payment request
-	DropAt uint     // threshold that triggers disconnect
+	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),
+				PayAt:  payAt,
-				DropAt: uint(dropAt),
+				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()
+	//defer swap.lock.Unlock()
-	swap.lock.Lock()
+	//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()
+	//defer swap.lock.Unlock()
-	swap.lock.Lock()
+	//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
 }
+*/

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)
+	instance := NewSwapProtocol()
-	if err != nil {
-		t.Fatal("Couldn't create Swap instance")
-	}
 	// wrapper function for servicenode to start the service
 	swapsvc := func(ctx *node.ServiceContext) (node.Service, error) {
 		return &API{
-			Swap: instance,
+			SwapProtocol: instance,
 		}, nil
 	}
 

swarm/swarm.go

@@ -208,10 +208,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 {
@@ -369,12 +371,14 @@ func (self *Swarm) Start(srv *p2p.Server) error {
 	}
 	log.Info(fmt.Sprintf("Swarm network started on bzz address: %x", self.bzz.Hive.Overlay.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)