internal/peers.go - release.projects.messenger - Git at Google

 // Copyright 2016 The Vanadium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.

 package internal

 import (
 	"fmt"
 	"io"
 	"math/rand"
 	"sort"
 	"sync"
 	"time"

 	"v.io/v23/context"
 	"v.io/v23/discovery"
 	"v.io/v23/naming"
 	"v.io/v23/options"
 	"v.io/v23/rpc"
 	"v.io/v23/security/access"
 	"v.io/v23/verror"

 	"v.io/x/ref/lib/stats/counter"

 	"messenger/ifc"
 )

 func startPeerManager(ctx *context.T, id string, updateChan <-chan discovery.Update, ps *PubSub, store MessengerStorage, params Params, counters *Counters) *peerManager {
 	pm := &peerManager{
 		id:       id,
 		store:    store,
 		params:   params,
 		peers:    make(map[string]*peer),
 		ps:       ps,
 		counters: counters,
 		done:     make(chan struct{}),
 	}
 	go pm.loop(ctx, updateChan)
 	return pm
 }

 type peerManager struct {
 	mu       sync.Mutex
 	id       string
 	store    MessengerStorage
 	params   Params
 	peers    map[string]*peer
 	ps       *PubSub
 	counters *Counters
 	done     chan struct{}
 }

 func (pm *peerManager) debugString() string {
 	pm.mu.Lock()
 	defer pm.mu.Unlock()
 	active := []string{}
 	for _, p := range pm.peers {
 		if p.ctx != nil {
 			active = append(active, p.debugString())
 		}
 	}
 	return fmt.Sprintf("PeerManager ID:%s MaxActivePeers:%d NumPeers:%d Active:%v",
 		pm.id, pm.params.MaxActivePeers, len(pm.peers), active)
 }

 func (pm *peerManager) loop(ctx *context.T, updateChan <-chan discovery.Update) {
 	var (
 		changed   bool
 		lastCheck time.Time
 	)
 	for {
 		select {
 		case update, ok := <-updateChan:
 			if !ok {
 				pm.removePeer(ctx, "")
 				close(pm.done)
 				return
 			}
 			pm.processDiscoveryUpdate(ctx, update)
 			changed = true
 		case <-time.After(time.Second):
 			// Updates tend to arrive in batches. Calling
 			// checkActivePeers on a timer makes its cost
 			// independent of the number of updates received in
 			// the interval.
 			if changed || time.Since(lastCheck) >= time.Minute {
 				pm.checkActivePeers(ctx)
 				changed = false
 				lastCheck = time.Now()
 			}
 		}
 	}
 }

 // processDiscoveryUpdate adds and removes peers as reported in the discovery
 // update.
 func (pm *peerManager) processDiscoveryUpdate(ctx *context.T, update discovery.Update) {
 	pm.mu.Lock()
 	defer pm.mu.Unlock()
 	id := update.Id().String()
 	if update.IsLost() {
 		ctx.Infof("%s lost peer %s", pm.id, id)
 		pm.removePeer(ctx, id)
 		return
 	}
 	p, exists := pm.peers[id]
 	if !exists {
 		p = &peer{
 			peerId:   id,
 			localId:  pm.id,
 			store:    pm.store,
 			params:   pm.params,
 			counters: pm.counters,
 		}
 		p.errors = counter.New()
 		pm.peers[id] = p
 		pm.counters.numPeers.Incr(1)
 		ctx.Infof("%s has new peer %s", pm.id, id)
 	}
 	me := &naming.MountEntry{}
 	for _, a := range update.Addresses() {
 		me.Servers = append(me.Servers, naming.MountedServer{Server: a})
 	}
 	p.setMountEntry(me)
 }

 // checkActivePeers decides which peers this node should talk to.
 //
 // The MaxActivePeers parameter controls how many peers this node should talk
 // to. If MaxActivePeers >= the number of peers, this node can talk to all of
 // them.
 //
 // When MaxActivePeers is less than the number of peers, we need to choose which
 // peers to talk to such that message propagation to all the peers is likely.
 //
 // Each node has a unique ID, which is known by all of its peers. So peers can
 // be sorted and collectively treated as a ring.
 //
 // Each node then picks N peers to talk to, where N <= MaxActivePeers. These
 // N peers are always the first N power of 2 neighbors to the right, i.e.
 //
 // Ring: A->B->C->D->E->F->G->H->I->J->K->L->M (->A)
 //
 // If N=1, Node C talks to D (distance 1)
 // If N=2, Node C talks to D, and E (distance 2)
 // If N=3, Node C talks to D, E, and G (distance 4)
 // If N=4, Node C talks to D, E, G, and K (distance 8)
 // etc.
 // When a power of 2 falls on a neighbor that was already selected, the next
 // unselected neighbor is selected instead.
 func (pm *peerManager) checkActivePeers(ctx *context.T) {
 	pm.mu.Lock()
 	defer pm.mu.Unlock()
 	numPeers := len(pm.peers)
 	if pm.params.MaxActivePeers == 0 || numPeers <= pm.params.MaxActivePeers {
 		for _, p := range pm.peers {
 			pm.startPeer(ctx, p)
 		}
 		return
 	}

 	sortedPeers := make([]string, 0, numPeers)
 	for id := range pm.peers {
 		sortedPeers = append(sortedPeers, id)
 	}
 	sort.Strings(sortedPeers)

 	// Find this node's (would-be) position in the list.
 	offset := 0
 	for ; offset < numPeers; offset++ {
 		if pm.id < sortedPeers[offset] {
 			break
 		}
 	}

 	// Calculate which neighboring nodes to talk to.
 	const peerErrorThreshold = 10
 	bad := make([]bool, numPeers)
 	for i := 0; i < numPeers; i++ {
 		if pm.peers[sortedPeers[i]].errors.Delta1h() >= peerErrorThreshold {
 			bad[i] = true
 		}
 	}

 	active := make([]bool, numPeers)
 	for i := 0; i < len(active) && i < pm.params.MaxActivePeers; i++ {
 		idx := (offset + (1<<uint(i%32) - 1)) % numPeers
 		// Skip peers that are already selected or that are bad.
 		for j := 0; j < numPeers; j++ {
 			if !active[idx] && !bad[idx] {
 				break
 			}
 			idx = (idx + 1) % numPeers
 		}
 		// If we can't find a peer that isn't already selected or bad,
 		// just find one that isn't already selected.
 		for active[idx] {
 			idx = (idx + 1) % numPeers
 		}
 		active[idx] = true
 	}

 	for i := 0; i < numPeers; i++ {
 		p := pm.peers[sortedPeers[i]]
 		if active[i] {
 			pm.startPeer(ctx, p)
 		} else {
 			pm.stopPeer(ctx, p)
 		}
 	}
 }

 // startPeer starts sending messages to the peer.
 func (pm *peerManager) startPeer(ctx *context.T, p *peer) {
 	if p.ctx == nil {
 		ctx.Infof("start peer %s", p.peerId)
 		p.ctx, p.cancel = context.WithCancel(ctx)
 		p.queue = newMsgQueue(p.ctx)
 		p.psChan = pm.ps.Sub()
 		p.done = make(chan struct{})
 		go p.subLoop()
 		go p.msgLoop()
 	}
 }

 // stopPeer stops sending messages to the peer.
 func (pm *peerManager) stopPeer(ctx *context.T, p *peer) {
 	if p.ctx != nil {
 		ctx.Infof("stop peer %s", p.peerId)
 		pm.ps.Unsub(p.psChan)
 		p.cancel()
 		<-p.done
 		p.ctx = nil
 		p.cancel = nil
 		p.psChan = nil
 		p.queue = nil
 		p.done = nil
 	}
 }

 // removePeer removes the peer from the peer list, after stopping it.
 func (pm *peerManager) removePeer(ctx *context.T, id string) {
 	for _, p := range pm.peers {
 		delete(pm.peers, id)
 		pm.stopPeer(ctx, p)
 		pm.counters.numPeers.Incr(-1)
 	}
 }

 type peer struct {
 	ctx      *context.T
 	cancel   func()
 	localId  string
 	peerId   string
 	store    MessengerStorage
 	params   Params
 	queue    *MsgQueue
 	counters *Counters
 	psChan   chan *ifc.Message
 	done     chan struct{}
 	errors   *counter.Counter
 	// mu guards me
 	mu sync.Mutex
 	me *naming.MountEntry
 }

 func (p *peer) debugString() string {
 	l := 0
 	if p.queue != nil {
 		l = p.queue.Len()
 	}
 	return fmt.Sprintf("Peer ID:%s QueueLen:%d", p.peerId, l)
 }

 func (p *peer) setMountEntry(me *naming.MountEntry) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	p.me = me
 }

 func (p *peer) mountEntry() *naming.MountEntry {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	me := *p.me
 	return &me
 }

 func (p *peer) checkHops(ctx *context.T, msg *ifc.Message) error {
 	for _, h := range msg.Hops {
 		if h == p.peerId {
 			return ifc.NewErrAlreadySeen(ctx, msg.Id)
 		}
 	}
 	if p.params.MaxHops > 0 && len(msg.Hops) >= p.params.MaxHops {
 		return ifc.NewErrTooManyHops(ctx, int32(p.params.MaxHops))
 	}
 	return nil
 }

 func (p *peer) subLoop() {
 	ctx := p.ctx
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case msg := <-p.psChan:
 			if msg == nil {
 				return
 			}
 			if p.checkHops(ctx, msg) != nil {
 				continue
 			}
 			p.queue.Insert(msg.CreationTime, msg.Id)
 		}
 	}
 }

 func (p *peer) msgLoop() {
 	const maxBackoff = 5 * time.Minute
 	ctx := p.ctx
 	for attempt := uint(0); ; attempt++ {
 		if err := backoff(p.ctx, func() error { return p.diffAndEnqueue(p.ctx) }); err != nil {
 			p.errors.Incr(1)
 			delay := time.Duration((5+rand.Intn(5))<<attempt) * time.Second
 			if delay > maxBackoff {
 				delay = maxBackoff
 			}
 			select {
 			case <-p.ctx.Done():
 				close(p.done)
 				return
 			case <-time.After(delay):
 			}
 			continue
 		}
 		break
 	}
 	for {
 		var msgId string

 		select {
 		case <-ctx.Done():
 			close(p.done)
 			return
 		case msgId = <-p.queue.Pop():
 		}

 		pi := p.store.PeerInfo(ctx, msgId, p.peerId)
 		if pi.Sent || pi.FatalError {
 			continue
 		}
 		if time.Now().Before(pi.NextAttempt) {
 			p.queue.Insert(pi.NextAttempt, msgId)
 			continue
 		}
 		pi.NumAttempts++

 		switch err := p.pushMessage(ctx, msgId); {
 		case err == nil:
 			pi.Sent = true
 			p.counters.numMessagesSent.Incr(1)
 		case verror.ErrorID(err) == ifc.ErrAlreadySeen.ID:
 			pi.Sent = true
 		case verror.ErrorID(err) == ifc.ErrContentMismatch.ID:
 			pi.FatalError = true
 		case verror.ErrorID(err) == ifc.ErrExpired.ID:
 			pi.FatalError = true
 		case verror.ErrorID(err) == ifc.ErrInvalidSignature.ID:
 			pi.FatalError = true
 		case verror.ErrorID(err) == ifc.ErrMessageIdCollision.ID:
 			pi.FatalError = true
 		case verror.ErrorID(err) == ifc.ErrNoRoute.ID:
 			pi.FatalError = true
 		case verror.ErrorID(err) == ifc.ErrTooManyHops.ID:
 			pi.FatalError = true
 		case verror.ErrorID(err) == ifc.ErrTooBig.ID:
 			pi.FatalError = true
 		case verror.ErrorID(err) == access.ErrNoPermissions.ID:
 			pi.FatalError = true
 		case verror.ErrorID(err) == verror.ErrUnknownMethod.ID:
 			pi.FatalError = true
 		case verror.ErrorID(err) == verror.ErrBadArg.ID:
 			pi.FatalError = true
 		case verror.ErrorID(err) == verror.ErrNoAccess.ID:
 			pi.FatalError = true
 		case verror.ErrorID(err) == verror.ErrNotTrusted.ID:
 			pi.FatalError = true
 		case verror.ErrorID(err) == verror.ErrNoExist.ID:
 			pi.FatalError = true
 		default:
 			n := uint(pi.NumAttempts)
 			// This is (10 to 20) * 2^n sec.
 			backoff := time.Duration((10+rand.Intn(10))<<n) * time.Second
 			if backoff > maxBackoff {
 				backoff = maxBackoff
 			}
 			pi.NextAttempt = time.Now().Add(backoff)
 			p.queue.Insert(pi.NextAttempt, msgId)
 		}
 		if pi.FatalError || pi.NumAttempts > 2 {
 			p.errors.Incr(1)
 		}
 		if err := p.store.SetPeerInfo(ctx, msgId, p.peerId, pi); err != nil {
 			ctx.Infof("store.SetPeerInfo failed: %v", err)
 		}
 	}
 }

 func (p *peer) pushMessage(ctx *context.T, msgId string) error {
 	ctx.Infof("Pushing %s to %s", msgId, p.peerId)
 	msg, r, err := p.store.OpenRead(ctx, msgId)
 	if err != nil {
 		return err
 	}
 	defer r.Close()

 	if err := msg.Expired(ctx); err != nil {
 		return err
 	}
 	msg.Hops = append(msg.Hops, p.localId)

 	opts := []rpc.CallOpt{
 		options.Preresolved{p.mountEntry()},
 		options.ServerAuthorizer{p.params.RateAclOut},
 	}

 	offset := int64(0)
 	fPush := func() error {
 		return p.doPush(ctx, msg, offset, r, opts)
 	}
 push:
 	if err = backoff(ctx, fPush); verror.ErrorID(err) == ifc.ErrIncorrectOffset.ID {
 		fOffset := func() error {
 			var err error
 			offset, err = ifc.MessengerClient("").ResumeOffset(ctx, msg, opts...)
 			return err
 		}
 		if err = backoff(ctx, fOffset); err != nil {
 			return err
 		}
 		if _, err := r.Seek(offset, 0); err != nil {
 			return err
 		}
 		goto push
 	}
 	return err
 }

 func backoff(ctx *context.T, f func() error) (err error) {
 	for n := uint(0); ; n++ {
 		if err = f(); verror.Action(err) == verror.RetryBackoff {
 			const maxBackoff = time.Minute
 			// This is (100 to 200) * 2^n ms.
 			backoff := time.Duration((100+rand.Intn(100))<<n) * time.Millisecond
 			if backoff > maxBackoff {
 				backoff = maxBackoff
 			}
 			select {
 			case <-ctx.Done():
 			case <-time.After(backoff):
 			}
 			continue
 		}
 		return
 	}
 }

 func (p *peer) doPush(ctx *context.T, msg ifc.Message, offset int64, r io.Reader, opts []rpc.CallOpt) error {
 	call, err := ifc.MessengerClient("").Push(ctx, msg, offset, opts...)
 	if err != nil {
 		return err
 	}
 	buf := make([]byte, 2048)
 	for {
 		n, err := r.Read(buf)
 		if n > 0 {
 			if err := call.SendStream().Send(buf[:n]); err != nil {
 				return err
 			}
 			p.counters.numBytesSent.Incr(int64(n))
 		}
 		if err == io.EOF {
 			break
 		}
 		if err != nil {
 			return err
 		}
 	}
 	return call.Finish()
 }

 func (p *peer) diffAndEnqueue(ctx *context.T) error {
 	opts := []rpc.CallOpt{
 		options.Preresolved{p.mountEntry()},
 		options.ServerAuthorizer{p.params.RateAclOut},
 	}
 	mctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	ch, err := p.store.Manifest(mctx)
 	if err != nil {
 		return err
 	}

 	const batchSize = 256
 	type batchItem struct {
 		id string
 		pi PeerInfo
 	}
 	var call ifc.MessengerDiffClientCall
 	for {
 		batch := make([]string, 0, batchSize)
 		batchInfo := make([]batchItem, 0, cap(batch))
 		for i := 0; i < cap(batch); i++ {
 			var msg *ifc.Message
 			select {
 			case <-ctx.Done():
 				return verror.NewErrCanceled(ctx)
 			case msg = <-ch:
 			}
 			if msg == nil {
 				break
 			}
 			pi := p.store.PeerInfo(ctx, msg.Id, p.peerId)
 			if pi.Sent || pi.FatalError {
 				continue
 			}
 			if p.checkHops(ctx, msg) != nil {
 				continue
 			}
 			batch = append(batch, msg.Id)
 			if pi.NextAttempt.IsZero() {
 				// Make an effort to deliver the messages in the
 				// order they were created.
 				pi.NextAttempt = msg.CreationTime
 			}
 			batchInfo = append(batchInfo, batchItem{msg.Id, pi})
 		}
 		if len(batch) == 0 {
 			break
 		}
 		if call == nil {
 			if call, err = ifc.MessengerClient("").Diff(ctx, opts...); err != nil {
 				return err
 			}
 		}
 		if err := call.SendStream().Send(batch); err != nil {
 			return err
 		}
 		if !call.RecvStream().Advance() {
 			return verror.New(verror.ErrBadProtocol, ctx)
 		}
 		responses := call.RecvStream().Value()
 		if len(batch) != len(responses) {
 			return verror.New(verror.ErrBadProtocol, ctx)
 		}
 		for i, hasIt := range responses {
 			id, pi := batchInfo[i].id, batchInfo[i].pi
 			if hasIt {
 				pi.Sent = true
 				if err := p.store.SetPeerInfo(ctx, id, p.peerId, pi); err != nil {
 					ctx.Infof("store.SetPeerInfo failed: %v", err)
 				}
 				continue
 			}
 			p.queue.Insert(pi.NextAttempt, id)
 		}
 	}
 	if call == nil {
 		return nil
 	}
 	if err := call.SendStream().Close(); err != nil {
 		return err
 	}
 	return call.Finish()
 }
	// Copyright 2016 The Vanadium Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style
	// license that can be found in the LICENSE file.

	package internal

	import (
	"fmt"
	"io"
	"math/rand"
	"sort"
	"sync"
	"time"

	"v.io/v23/context"
	"v.io/v23/discovery"
	"v.io/v23/naming"
	"v.io/v23/options"
	"v.io/v23/rpc"
	"v.io/v23/security/access"
	"v.io/v23/verror"

	"v.io/x/ref/lib/stats/counter"

	"messenger/ifc"
	)

	func startPeerManager(ctx context.T, id string, updateChan <-chan discovery.Update, ps PubSub, store MessengerStorage, params Params, counters Counters) peerManager {
	pm := &peerManager{
	id: id,
	store: store,
	params: params,
	peers: make(map[string]*peer),
	ps: ps,
	counters: counters,
	done: make(chan struct{}),
	}
	go pm.loop(ctx, updateChan)
	return pm
	}

	type peerManager struct {
	mu sync.Mutex
	id string
	store MessengerStorage
	params Params
	peers map[string]*peer
	ps *PubSub
	counters *Counters
	done chan struct{}
	}

	func (pm *peerManager) debugString() string {
	pm.mu.Lock()
	defer pm.mu.Unlock()
	active := []string{}
	for _, p := range pm.peers {
	if p.ctx != nil {
	active = append(active, p.debugString())
	}
	}
	return fmt.Sprintf("PeerManager ID:%s MaxActivePeers:%d NumPeers:%d Active:%v",
	pm.id, pm.params.MaxActivePeers, len(pm.peers), active)
	}

	func (pm peerManager) loop(ctx context.T, updateChan <-chan discovery.Update) {
	var (
	changed bool
	lastCheck time.Time
	)
	for {
	select {
	case update, ok := <-updateChan:
	if !ok {
	pm.removePeer(ctx, "")
	close(pm.done)
	return
	}
	pm.processDiscoveryUpdate(ctx, update)
	changed = true
	case <-time.After(time.Second):
	// Updates tend to arrive in batches. Calling
	// checkActivePeers on a timer makes its cost
	// independent of the number of updates received in
	// the interval.
	if changed \|\| time.Since(lastCheck) >= time.Minute {
	pm.checkActivePeers(ctx)
	changed = false
	lastCheck = time.Now()
	}
	}
	}
	}

	// processDiscoveryUpdate adds and removes peers as reported in the discovery
	// update.
	func (pm peerManager) processDiscoveryUpdate(ctx context.T, update discovery.Update) {
	pm.mu.Lock()
	defer pm.mu.Unlock()
	id := update.Id().String()
	if update.IsLost() {
	ctx.Infof("%s lost peer %s", pm.id, id)
	pm.removePeer(ctx, id)
	return
	}
	p, exists := pm.peers[id]
	if !exists {
	p = &peer{
	peerId: id,
	localId: pm.id,
	store: pm.store,
	params: pm.params,
	counters: pm.counters,
	}
	p.errors = counter.New()
	pm.peers[id] = p
	pm.counters.numPeers.Incr(1)
	ctx.Infof("%s has new peer %s", pm.id, id)
	}
	me := &naming.MountEntry{}
	for _, a := range update.Addresses() {
	me.Servers = append(me.Servers, naming.MountedServer{Server: a})
	}
	p.setMountEntry(me)
	}

	// checkActivePeers decides which peers this node should talk to.
	//
	// The MaxActivePeers parameter controls how many peers this node should talk
	// to. If MaxActivePeers >= the number of peers, this node can talk to all of
	// them.
	//
	// When MaxActivePeers is less than the number of peers, we need to choose which
	// peers to talk to such that message propagation to all the peers is likely.
	//
	// Each node has a unique ID, which is known by all of its peers. So peers can
	// be sorted and collectively treated as a ring.
	//
	// Each node then picks N peers to talk to, where N <= MaxActivePeers. These
	// N peers are always the first N power of 2 neighbors to the right, i.e.
	//
	// Ring: A->B->C->D->E->F->G->H->I->J->K->L->M (->A)
	//
	// If N=1, Node C talks to D (distance 1)
	// If N=2, Node C talks to D, and E (distance 2)
	// If N=3, Node C talks to D, E, and G (distance 4)
	// If N=4, Node C talks to D, E, G, and K (distance 8)
	// etc.
	// When a power of 2 falls on a neighbor that was already selected, the next
	// unselected neighbor is selected instead.
	func (pm peerManager) checkActivePeers(ctx context.T) {
	pm.mu.Lock()
	defer pm.mu.Unlock()
	numPeers := len(pm.peers)
	if pm.params.MaxActivePeers == 0 \|\| numPeers <= pm.params.MaxActivePeers {
	for _, p := range pm.peers {
	pm.startPeer(ctx, p)
	}
	return
	}

	sortedPeers := make([]string, 0, numPeers)
	for id := range pm.peers {
	sortedPeers = append(sortedPeers, id)
	}
	sort.Strings(sortedPeers)

	// Find this node's (would-be) position in the list.
	offset := 0
	for ; offset < numPeers; offset++ {
	if pm.id < sortedPeers[offset] {
	break
	}
	}

	// Calculate which neighboring nodes to talk to.
	const peerErrorThreshold = 10
	bad := make([]bool, numPeers)
	for i := 0; i < numPeers; i++ {
	if pm.peers[sortedPeers[i]].errors.Delta1h() >= peerErrorThreshold {
	bad[i] = true
	}
	}

	active := make([]bool, numPeers)
	for i := 0; i < len(active) && i < pm.params.MaxActivePeers; i++ {
	idx := (offset + (1<<uint(i%32) - 1)) % numPeers
	// Skip peers that are already selected or that are bad.
	for j := 0; j < numPeers; j++ {
	if !active[idx] && !bad[idx] {
	break
	}
	idx = (idx + 1) % numPeers
	}
	// If we can't find a peer that isn't already selected or bad,
	// just find one that isn't already selected.
	for active[idx] {
	idx = (idx + 1) % numPeers
	}
	active[idx] = true
	}

	for i := 0; i < numPeers; i++ {
	p := pm.peers[sortedPeers[i]]
	if active[i] {
	pm.startPeer(ctx, p)
	} else {
	pm.stopPeer(ctx, p)
	}
	}
	}

	// startPeer starts sending messages to the peer.
	func (pm peerManager) startPeer(ctx context.T, p *peer) {
	if p.ctx == nil {
	ctx.Infof("start peer %s", p.peerId)
	p.ctx, p.cancel = context.WithCancel(ctx)
	p.queue = newMsgQueue(p.ctx)
	p.psChan = pm.ps.Sub()
	p.done = make(chan struct{})
	go p.subLoop()
	go p.msgLoop()
	}
	}

	// stopPeer stops sending messages to the peer.
	func (pm peerManager) stopPeer(ctx context.T, p *peer) {
	if p.ctx != nil {
	ctx.Infof("stop peer %s", p.peerId)
	pm.ps.Unsub(p.psChan)
	p.cancel()
	<-p.done
	p.ctx = nil
	p.cancel = nil
	p.psChan = nil
	p.queue = nil
	p.done = nil
	}
	}

	// removePeer removes the peer from the peer list, after stopping it.
	func (pm peerManager) removePeer(ctx context.T, id string) {
	for _, p := range pm.peers {
	delete(pm.peers, id)
	pm.stopPeer(ctx, p)
	pm.counters.numPeers.Incr(-1)
	}
	}

	type peer struct {
	ctx *context.T
	cancel func()
	localId string
	peerId string
	store MessengerStorage
	params Params
	queue *MsgQueue
	counters *Counters
	psChan chan *ifc.Message
	done chan struct{}
	errors *counter.Counter
	// mu guards me
	mu sync.Mutex
	me *naming.MountEntry
	}

	func (p *peer) debugString() string {
	l := 0
	if p.queue != nil {
	l = p.queue.Len()
	}
	return fmt.Sprintf("Peer ID:%s QueueLen:%d", p.peerId, l)
	}

	func (p peer) setMountEntry(me naming.MountEntry) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.me = me
	}

	func (p peer) mountEntry() naming.MountEntry {
	p.mu.Lock()
	defer p.mu.Unlock()
	me := *p.me
	return &me
	}

	func (p peer) checkHops(ctx context.T, msg *ifc.Message) error {
	for _, h := range msg.Hops {
	if h == p.peerId {
	return ifc.NewErrAlreadySeen(ctx, msg.Id)
	}
	}
	if p.params.MaxHops > 0 && len(msg.Hops) >= p.params.MaxHops {
	return ifc.NewErrTooManyHops(ctx, int32(p.params.MaxHops))
	}
	return nil
	}

	func (p *peer) subLoop() {
	ctx := p.ctx
	for {
	select {
	case <-ctx.Done():
	return
	case msg := <-p.psChan:
	if msg == nil {
	return
	}
	if p.checkHops(ctx, msg) != nil {
	continue
	}
	p.queue.Insert(msg.CreationTime, msg.Id)
	}
	}
	}

	func (p *peer) msgLoop() {
	const maxBackoff = 5 * time.Minute
	ctx := p.ctx
	for attempt := uint(0); ; attempt++ {
	if err := backoff(p.ctx, func() error { return p.diffAndEnqueue(p.ctx) }); err != nil {
	p.errors.Incr(1)
	delay := time.Duration((5+rand.Intn(5))<<attempt) * time.Second
	if delay > maxBackoff {
	delay = maxBackoff
	}
	select {
	case <-p.ctx.Done():
	close(p.done)
	return
	case <-time.After(delay):
	}
	continue
	}
	break
	}
	for {
	var msgId string

	select {
	case <-ctx.Done():
	close(p.done)
	return
	case msgId = <-p.queue.Pop():
	}

	pi := p.store.PeerInfo(ctx, msgId, p.peerId)
	if pi.Sent \|\| pi.FatalError {
	continue
	}
	if time.Now().Before(pi.NextAttempt) {
	p.queue.Insert(pi.NextAttempt, msgId)
	continue
	}
	pi.NumAttempts++

	switch err := p.pushMessage(ctx, msgId); {
	case err == nil:
	pi.Sent = true
	p.counters.numMessagesSent.Incr(1)
	case verror.ErrorID(err) == ifc.ErrAlreadySeen.ID:
	pi.Sent = true
	case verror.ErrorID(err) == ifc.ErrContentMismatch.ID:
	pi.FatalError = true
	case verror.ErrorID(err) == ifc.ErrExpired.ID:
	pi.FatalError = true
	case verror.ErrorID(err) == ifc.ErrInvalidSignature.ID:
	pi.FatalError = true
	case verror.ErrorID(err) == ifc.ErrMessageIdCollision.ID:
	pi.FatalError = true
	case verror.ErrorID(err) == ifc.ErrNoRoute.ID:
	pi.FatalError = true
	case verror.ErrorID(err) == ifc.ErrTooManyHops.ID:
	pi.FatalError = true
	case verror.ErrorID(err) == ifc.ErrTooBig.ID:
	pi.FatalError = true
	case verror.ErrorID(err) == access.ErrNoPermissions.ID:
	pi.FatalError = true
	case verror.ErrorID(err) == verror.ErrUnknownMethod.ID:
	pi.FatalError = true
	case verror.ErrorID(err) == verror.ErrBadArg.ID:
	pi.FatalError = true
	case verror.ErrorID(err) == verror.ErrNoAccess.ID:
	pi.FatalError = true
	case verror.ErrorID(err) == verror.ErrNotTrusted.ID:
	pi.FatalError = true
	case verror.ErrorID(err) == verror.ErrNoExist.ID:
	pi.FatalError = true
	default:
	n := uint(pi.NumAttempts)
	// This is (10 to 20) * 2^n sec.
	backoff := time.Duration((10+rand.Intn(10))<<n) * time.Second
	if backoff > maxBackoff {
	backoff = maxBackoff
	}
	pi.NextAttempt = time.Now().Add(backoff)
	p.queue.Insert(pi.NextAttempt, msgId)
	}
	if pi.FatalError \|\| pi.NumAttempts > 2 {
	p.errors.Incr(1)
	}
	if err := p.store.SetPeerInfo(ctx, msgId, p.peerId, pi); err != nil {
	ctx.Infof("store.SetPeerInfo failed: %v", err)
	}
	}
	}

	func (p peer) pushMessage(ctx context.T, msgId string) error {
	ctx.Infof("Pushing %s to %s", msgId, p.peerId)
	msg, r, err := p.store.OpenRead(ctx, msgId)
	if err != nil {
	return err
	}
	defer r.Close()

	if err := msg.Expired(ctx); err != nil {
	return err
	}
	msg.Hops = append(msg.Hops, p.localId)

	opts := []rpc.CallOpt{
	options.Preresolved{p.mountEntry()},
	options.ServerAuthorizer{p.params.RateAclOut},
	}

	offset := int64(0)
	fPush := func() error {
	return p.doPush(ctx, msg, offset, r, opts)
	}
	push:
	if err = backoff(ctx, fPush); verror.ErrorID(err) == ifc.ErrIncorrectOffset.ID {
	fOffset := func() error {
	var err error
	offset, err = ifc.MessengerClient("").ResumeOffset(ctx, msg, opts...)
	return err
	}
	if err = backoff(ctx, fOffset); err != nil {
	return err
	}
	if _, err := r.Seek(offset, 0); err != nil {
	return err
	}
	goto push
	}
	return err
	}

	func backoff(ctx *context.T, f func() error) (err error) {
	for n := uint(0); ; n++ {
	if err = f(); verror.Action(err) == verror.RetryBackoff {
	const maxBackoff = time.Minute
	// This is (100 to 200) * 2^n ms.
	backoff := time.Duration((100+rand.Intn(100))<<n) * time.Millisecond
	if backoff > maxBackoff {
	backoff = maxBackoff
	}
	select {
	case <-ctx.Done():
	case <-time.After(backoff):
	}
	continue
	}
	return
	}
	}

	func (p peer) doPush(ctx context.T, msg ifc.Message, offset int64, r io.Reader, opts []rpc.CallOpt) error {
	call, err := ifc.MessengerClient("").Push(ctx, msg, offset, opts...)
	if err != nil {
	return err
	}
	buf := make([]byte, 2048)
	for {
	n, err := r.Read(buf)
	if n > 0 {
	if err := call.SendStream().Send(buf[:n]); err != nil {
	return err
	}
	p.counters.numBytesSent.Incr(int64(n))
	}
	if err == io.EOF {
	break
	}
	if err != nil {
	return err
	}
	}
	return call.Finish()
	}

	func (p peer) diffAndEnqueue(ctx context.T) error {
	opts := []rpc.CallOpt{
	options.Preresolved{p.mountEntry()},
	options.ServerAuthorizer{p.params.RateAclOut},
	}
	mctx, cancel := context.WithCancel(ctx)
	defer cancel()
	ch, err := p.store.Manifest(mctx)
	if err != nil {
	return err
	}

	const batchSize = 256
	type batchItem struct {
	id string
	pi PeerInfo
	}
	var call ifc.MessengerDiffClientCall
	for {
	batch := make([]string, 0, batchSize)
	batchInfo := make([]batchItem, 0, cap(batch))
	for i := 0; i < cap(batch); i++ {
	var msg *ifc.Message
	select {
	case <-ctx.Done():
	return verror.NewErrCanceled(ctx)
	case msg = <-ch:
	}
	if msg == nil {
	break
	}
	pi := p.store.PeerInfo(ctx, msg.Id, p.peerId)
	if pi.Sent \|\| pi.FatalError {
	continue
	}
	if p.checkHops(ctx, msg) != nil {
	continue
	}
	batch = append(batch, msg.Id)
	if pi.NextAttempt.IsZero() {
	// Make an effort to deliver the messages in the
	// order they were created.
	pi.NextAttempt = msg.CreationTime
	}
	batchInfo = append(batchInfo, batchItem{msg.Id, pi})
	}
	if len(batch) == 0 {
	break
	}
	if call == nil {
	if call, err = ifc.MessengerClient("").Diff(ctx, opts...); err != nil {
	return err
	}
	}
	if err := call.SendStream().Send(batch); err != nil {
	return err
	}
	if !call.RecvStream().Advance() {
	return verror.New(verror.ErrBadProtocol, ctx)
	}
	responses := call.RecvStream().Value()
	if len(batch) != len(responses) {
	return verror.New(verror.ErrBadProtocol, ctx)
	}
	for i, hasIt := range responses {
	id, pi := batchInfo[i].id, batchInfo[i].pi
	if hasIt {
	pi.Sent = true
	if err := p.store.SetPeerInfo(ctx, id, p.peerId, pi); err != nil {
	ctx.Infof("store.SetPeerInfo failed: %v", err)
	}
	continue
	}
	p.queue.Insert(pi.NextAttempt, id)
	}
	}
	if call == nil {
	return nil
	}
	if err := call.SendStream().Close(); err != nil {
	return err
	}
	return call.Finish()
	}