swarm/pushsync, swarm/pss: push sync implementation

Author: zelig

swarm/pss/pubsub.go

@@ -0,0 +1,41 @@
+package pss
+
+import "github.com/ethereum/go-ethereum/p2p"
+
+// PubSub implements the pusher.PubSub interface using pss
+type PubSub struct {
+	pss  *Pss // pss
+	prox bool // determines if pss send should use neighbourhood addressing
+}
+
+// NewPubSub creates a new PubSub
+func NewPubSub(p *Pss, prox bool) *PubSub {
+	return &PubSub{
+		pss:  p,
+		prox: prox,
+	}
+}
+
+// BaseAddr returns Kademlia base address
+func (p *PubSub) BaseAddr() []byte {
+	return p.pss.Kademlia.BaseAddr()
+}
+
+// Register registers a handler
+func (p *PubSub) Register(topic string, handler func(msg []byte, p *p2p.Peer) error) func() {
+	f := func(msg []byte, peer *p2p.Peer, _ bool, _ string) error {
+		return handler(msg, peer)
+	}
+	h := NewHandler(f).WithRaw()
+	if p.prox {
+		h = h.WithProxBin()
+	}
+	pt := BytesToTopic([]byte(topic))
+	return p.pss.Register(&pt, h)
+}
+
+// Send sends a message using pss SendRaw
+func (p *PubSub) Send(to []byte, topic string, msg []byte) error {
+	pt := BytesToTopic([]byte(topic))
+	return p.pss.SendRaw(PssAddress(to), pt, msg)
+}

swarm/storage/pushsync/protocol.go

@@ -0,0 +1,71 @@
+package pushsync
+
+import (
+	"crypto/rand"
+
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/rlp"
+)
+
+const (
+	pssChunkTopic   = "PUSHSYNC_CHUNKS"   // pss topic for chunks
+	pssReceiptTopic = "PUSHSYNC_RECEIPTS" // pss topic for statement of custody receipts
+)
+
+// PubSub is a Postal Service interface needed to send/receive chunks and receipts for push syncing
+type PubSub interface {
+	Register(topic string, handler func(msg []byte, p *p2p.Peer) error) func()
+	Send(to []byte, topic string, msg []byte) error
+	BaseAddr() []byte
+}
+
+// chunkMsg is the message construct to send chunks to their local neighbourhood
+type chunkMsg struct {
+	Addr   []byte // chunk address
+	Data   []byte // chunk data
+	Origin []byte // originator
+	Nonce  []byte // nonce to make multiple instances of send immune to deduplication cache
+}
+
+// receiptMsg is a statement of custody response to receiving a push-synced chunk
+// it is currently a notification only (contains no proof) sent to the originator
+// Nonce is there to make multiple responses immune to deduplication cache
+type receiptMsg struct {
+	Addr  []byte
+	Nonce []byte
+}
+
+func decodeChunkMsg(msg []byte) (*chunkMsg, error) {
+	var chmsg chunkMsg
+	err := rlp.DecodeBytes(msg, &chmsg)
+	if err != nil {
+		return nil, err
+	}
+	return &chmsg, nil
+}
+
+func decodeReceiptMsg(msg []byte) (*receiptMsg, error) {
+	var rmsg receiptMsg
+	err := rlp.DecodeBytes(msg, &rmsg)
+	if err != nil {
+		return nil, err
+	}
+	return &rmsg, nil
+}
+
+// newNonce creates a random nonce;
+// even without POC it is important otherwise resending a chunk is deduplicated by pss
+func newNonce() []byte {
+	buf := make([]byte, 32)
+	t := 0
+	for t < len(buf) {
+		n, _ := rand.Read(buf[t:])
+		t += n
+	}
+	return buf
+}
+
+func label(b []byte) string {
+	return hexutil.Encode(b[:8])
+}

swarm/storage/pushsync/pusher.go

@@ -0,0 +1,228 @@
+package pushsync
+
+import (
+	"context"
+	"time"
+
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/rlp"
+	"github.com/ethereum/go-ethereum/swarm/chunk"
+	"github.com/ethereum/go-ethereum/swarm/log"
+	"github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+// DB interface
+type DB interface {
+	// subscribe to chunk to be push synced - iterates from earliest to newest
+	SubscribePush(context.Context) (<-chan storage.Chunk, func())
+	// called to set a chunk as synced - and allow it to be garbage collected
+	// TODO this should take ... last argument to delete many in one batch
+	Set(context.Context, chunk.ModeSet, storage.Address) error
+}
+
+// Pusher takes care of the push syncing
+type Pusher struct {
+	store    DB                     // localstore DB
+	tags     *chunk.Tags            // tags to update counts
+	quit     chan struct{}          // channel to signal quitting on all loops
+	pushed   map[string]*pushedItem // cache of items push-synced
+	receipts chan chunk.Address     // channel to receive receipts
+	ps       PubSub                 // PubSub interface to send chunks and receive receipts
+}
+
+var (
+	retryInterval = 1 * time.Second // seconds to wait before retry sync
+)
+
+// pushedItem captures the info needed for the pusher about a chunk during the
+// push-sync--receipt roundtrip
+type pushedItem struct {
+	tag    *chunk.Tag // tag for the chunk
+	sentAt time.Time  // time last pushed
+	synced bool       // set when
+}
+
+// New contructs a Pusher and starts up the push sync protocol
+func New(store DB, ps PubSub, tags *chunk.Tags) *Pusher {
+	p := &Pusher{
+		store:    store,
+		tags:     tags,
+		quit:     make(chan struct{}),
+		pushed:   make(map[string]*pushedItem),
+		receipts: make(chan chunk.Address),
+		ps:       ps,
+	}
+	go p.sync()
+	return p
+}
+
+// Close closes the pusher
+func (p *Pusher) Close() {
+	close(p.quit)
+}
+
+// sync starts a forever loop that pushes chunks to their neighbourhood
+// and receives receipts for them
+// chunks that are not acknowledged with a receipt are retried
+// not earlier than retryInterval after they were last  pushed
+// the routine also updates counts of states on a tag in order
+// to monitor the proportion of saved, sent and synced chunks of
+// a file or collection
+func (p *Pusher) sync() {
+	var chunks <-chan chunk.Chunk
+	var cancel, stop func()
+	var ctx context.Context
+	var synced []storage.Address
+
+	// timer
+	timer := time.NewTimer(0)
+	defer timer.Stop()
+
+	// register handler for pssReceiptTopic on pss pubsub
+	deregister := p.ps.Register(pssReceiptTopic, func(msg []byte, _ *p2p.Peer) error {
+		return p.handleReceiptMsg(msg)
+	})
+	defer deregister()
+
+	for {
+		select {
+
+		// handle incoming chunks
+		case ch, more := <-chunks:
+			// if no more, set to nil and wait for timer
+			if !more {
+				chunks = nil
+				continue
+			}
+			// if no need to sync this chunk then continue
+			if !p.needToSync(ch) {
+				continue
+			}
+			// send the chunk and ignore the error
+			if err := p.sendChunkMsg(ch); err != nil {
+				log.Warn("error sending chunk", "addr", ch.Address(), "err", err)
+			}
+
+			// handle incoming receipts
+		case addr := <-p.receipts:
+			log.Warn("synced", "addr", addr)
+			// ignore if already received receipt
+			item, found := p.pushed[addr.Hex()]
+			if !found {
+				log.Debug("not wanted or already got... ignore", "addr", addr)
+				continue
+			}
+			if item.synced {
+				log.Debug("just synced... ignore", "addr", addr)
+				continue
+			}
+			// collect synced addresses
+			synced = append(synced, addr)
+			// set synced flag
+			item.synced = true
+			// increment synced count for the tag if exists
+			if item.tag != nil {
+				item.tag.Inc(chunk.StateSynced)
+				log.Warn("tag", "uid", item.tag.Uid, "synced", item.tag.Get(chunk.StateSynced))
+			}
+			log.Warn("synced", "addr", addr)
+
+			// retry interval timer triggers starting from new
+		case <-timer.C:
+			log.Debug("timer")
+			// TODO: implement some smart retry strategy relying on sent/synced ratio change
+			// if subscribe was running, stop it
+			if stop != nil {
+				stop()
+			}
+			for _, addr := range synced {
+				// set chunk status to synced, insert to db GC index
+				if err := p.store.Set(context.Background(), chunk.ModeSetSync, addr); err != nil {
+					log.Warn("error setting chunk to synced", "addr", addr, "err", err)
+					continue
+				}
+				delete(p.pushed, addr.Hex())
+			}
+			synced = nil
+
+			// and start from the beginning
+			ctx, cancel = context.WithCancel(context.Background())
+			chunks, stop = p.store.SubscribePush(ctx)
+			// reset timer to go off after retryInterval
+			timer.Reset(retryInterval)
+
+		case <-p.quit:
+			// if there was a subscription, cancel it
+			if cancel != nil {
+				cancel()
+			}
+			return
+		}
+	}
+}
+
+// handleReceiptMsg id a handler for pssReceiptTopic that
+// - deserialises receiptMsg and
+// - sends the receipted address on a channel
+func (p *Pusher) handleReceiptMsg(msg []byte) error {
+	receipt, err := decodeReceiptMsg(msg)
+	if err != nil {
+		return err
+	}
+	log.Debug("Handler", "receipt", label(receipt.Addr), "self", label(p.ps.BaseAddr()))
+	select {
+	case p.receipts <- receipt.Addr:
+	case <-p.quit:
+	}
+	return nil
+}
+
+// sendChunkMsg sends chunks to their destination
+// using the PubSub interface Send method (e.g., pss neighbourhood addressing)
+func (p *Pusher) sendChunkMsg(ch chunk.Chunk) error {
+	cmsg := &chunkMsg{
+		Origin: p.ps.BaseAddr(),
+		Addr:   ch.Address()[:],
+		Data:   ch.Data(),
+		Nonce:  newNonce(),
+	}
+	msg, err := rlp.EncodeToBytes(cmsg)
+	if err != nil {
+		return err
+	}
+	log.Warn("send chunk", "addr", label(ch.Address()), "self", label(p.ps.BaseAddr()))
+	return p.ps.Send(ch.Address()[:], pssChunkTopic, msg)
+}
+
+// needToSync checks if a chunk needs to be push-synced:
+// * if not sent yet OR
+// * if sent but more then retryInterval ago, so need resend
+ func (p *Pusher) needToSync(ch chunk.Chunk) bool {
+	item, found := p.pushed[ch.Address().Hex()]
+	// has been pushed already
+	if found {
+		// has synced already since subscribe called
+		if item.synced {
+			return false
+		}
+		// too early to retry
+		if item.sentAt.Add(retryInterval).After(time.Now()) {
+			return false
+		}
+		// first time encountered
+	} else {
+		// remember item
+		tag, _ := p.tags.Get(ch.Tag())
+		log.Warn("tag", "uid from db", ch.Tag())
+		item = &pushedItem{
+			tag: tag,
+		}
+		if item.tag != nil {
+			item.tag.Inc(chunk.StateSent)
+		}
+		log.Warn("tag", "uid", tag.Uid, "sent", tag.Get(chunk.StateSent))
+		p.pushed[ch.Address().Hex()] = item
+	}
+	item.sentAt = time.Now()
+	return true
+}