multipeer

Documentation

Index

• Constants
• type MultiPeerReconciler
  • func NewMultiPeerReconciler(logger *zap.Logger, cfg MultiPeerReconcilerConfig, syncBase SyncBase, ...) *MultiPeerReconciler
  • func (mpr *MultiPeerReconciler) Run(ctx context.Context, kickCh chan struct{}) error
  • func (mpr *MultiPeerReconciler) Synced() bool
• type MultiPeerReconcilerConfig
  • func DefaultConfig() MultiPeerReconcilerConfig
  • func (cfg *MultiPeerReconcilerConfig) Validate(logger *zap.Logger) bool
• type PairwiseSyncer
• type SetSyncBase
  • func NewSetSyncBase(ps PairwiseSyncer, os rangesync.OrderedSet, handler SyncKeyHandler) *SetSyncBase
  • func (ssb *SetSyncBase) Count() (int, error)
  • func (ssb *SetSyncBase) Probe(ctx context.Context, p p2p.Peer) (pr rangesync.ProbeResult, err error)
  • func (ssb *SetSyncBase) Serve(ctx context.Context, p p2p.Peer, stream io.ReadWriter) error
  • func (ssb *SetSyncBase) Sync(ctx context.Context, p p2p.Peer, x, y rangesync.KeyBytes) error
• type SyncBase
• type SyncKeyHandler

Constants

const (
	Protocol = "sync/2"
)

Types

type MultiPeerReconciler

type MultiPeerReconciler struct {
	// contains filtered or unexported fields
}

MultiPeerReconciler reconcilies the local set against multiple remote sets.

func NewMultiPeerReconciler

func NewMultiPeerReconciler(
	logger *zap.Logger,
	cfg MultiPeerReconcilerConfig,
	syncBase SyncBase,
	peers *peers.Peers,
	keyLen, maxDepth int,
) *MultiPeerReconciler

NewMultiPeerReconciler creates a new MultiPeerReconciler.

func (*MultiPeerReconciler) Run

func (mpr *MultiPeerReconciler) Run(ctx context.Context, kickCh chan struct{}) error

Run runs the MultiPeerReconciler.

func (*MultiPeerReconciler) Synced

func (mpr *MultiPeerReconciler) Synced() bool

Synced returns true if the node is considered synced, that is, the specified number of full syncs has happened within the specified duration of time.

type MultiPeerReconcilerConfig

type MultiPeerReconcilerConfig struct {
	// Number of peers to pick for synchronization.
	// Synchronization will still happen if fewer peers are available.
	SyncPeerCount uint `mapstructure:"sync-peer-count"`
	// Minimum number of peers for the split sync to happen.
	MinSplitSyncPeers uint `mapstructure:"min-split-sync-peers"`
	// Minimum number of items that a peer must have to be eligible for split sync
	// (subrange-per-peer).
	MinSplitSyncCount uint `mapstructure:"min-split-sync-count"`
	// Maximum approximate size of symmetric difference between the local set and the
	// remote one for the sets to be considered "mostly in sync", so that full sync is
	// preferred to split sync.
	MaxFullDiff uint `mapstructure:"max-full-diff"`
	// Maximum number of items that a peer can have less than the local set for it to
	// be considered for synchronization.
	MaxSyncDiff uint `mapstructure:"max-sync-diff"`
	// Minimum fraction (0..1) of "mostly synced" peers starting with which full sync
	// is used instead of split sync.
	MinCompleteFraction float64 `mapstructure:"min-complete-fraction"`
	// Interval between syncs.
	SyncInterval time.Duration `mapstructure:"sync-interval"`
	// Interval spread factor for split sync.
	// The actual interval will be SyncInterval * (1 + (random[0..2]*SplitSyncIntervalSpread-1)).
	// So, if you want the actual interval to be with the range of SplitInterval ± 25%,
	// set this to 0.25.
	SyncIntervalSpread float64 `mapstructure:"sync-interval-spread"`
	// Interval between retries after a failed sync.
	RetryInterval time.Duration `mapstructure:"retry-interval"`
	// Interval between rechecking for peers after no synchronization peers were
	// found.
	NoPeersRecheckInterval time.Duration `mapstructure:"no-peers-recheck-interval"`
	// Grace period for split sync peers.
	// If a peer doesn't complete syncing its range within the specified duration
	// during split sync, its range is assigned additionally to another quicker
	// peer. The sync against the "slow" peer is NOT stopped immediately after that.
	SplitSyncGracePeriod time.Duration `mapstructure:"split-sync-grace-period"`
	// Minimum number of full syncs that must have happened within the
	// fullSyncednessPeriod for the node to be considered fully synced
	MinFullSyncednessCount uint `mapstructure:"min-full-syncedness-count"`
	// Duration within which the minimum number of full syncs must have happened for
	// the node to be considered fully synced.
	FullSyncednessPeriod time.Duration `mapstructure:"full-syncedness-count"`
}

MultiPeerReconcilerConfig contains the configuration for a MultiPeerReconciler.

func DefaultConfig

func DefaultConfig() MultiPeerReconcilerConfig

DefaultConfig returns the default configuration for the MultiPeerReconciler.

func (*MultiPeerReconcilerConfig) Validate

func (cfg *MultiPeerReconcilerConfig) Validate(logger *zap.Logger) bool

type PairwiseSyncer

type PairwiseSyncer interface {
	// Probe probes the peer using the specified range, to check how different the
	// peer's set is from the local set.
	Probe(
		ctx context.Context,
		peer p2p.Peer,
		os rangesync.OrderedSet,
		x, y rangesync.KeyBytes,
	) (rangesync.ProbeResult, error)
	// Sync synchronizes the set with the peer using the specified range.
	Sync(
		ctx context.Context,
		peer p2p.Peer,
		os rangesync.OrderedSet,
		x, y rangesync.KeyBytes,
	) error
	// Serve serves an incoming synchronization request.
	Serve(context context.Context, stream io.ReadWriter, os rangesync.OrderedSet) error
}

PairwiseSyncer is used to probe a peer or sync against a single peer. It does not contain a copy of the set.

type SetSyncBase

type SetSyncBase struct {
	// contains filtered or unexported fields
}

SetSyncBase is a synchronization base which holds the original OrderedSet. For each peer, a Syncer is derived from the base which is used to synchronize against that peer only. This way, there's no propagation of any keys for which the actual data has not been yet received and validated.

func NewSetSyncBase

func NewSetSyncBase(
	ps PairwiseSyncer,
	os rangesync.OrderedSet,
	handler SyncKeyHandler,
) *SetSyncBase

NewSetSyncBase creates a new SetSyncBase.

func (*SetSyncBase) Count

func (ssb *SetSyncBase) Count() (int, error)

Count implements SyncBase.

func (*SetSyncBase) Probe

func (ssb *SetSyncBase) Probe(ctx context.Context, p p2p.Peer) (pr rangesync.ProbeResult, err error)

Probe implements SyncBase.

func (*SetSyncBase) Serve

func (ssb *SetSyncBase) Serve(ctx context.Context, p p2p.Peer, stream io.ReadWriter) error

func (*SetSyncBase) Sync

func (ssb *SetSyncBase) Sync(ctx context.Context, p p2p.Peer, x, y rangesync.KeyBytes) error

type SyncBase

type SyncBase interface {
	// Count returns the number of items in the set.
	Count() (int, error)
	// Sync synchronizes the set with the peer.
	// It returns a sequence of new keys that were received from the peer and the
	// number of received items.
	Sync(ctx context.Context, p p2p.Peer, x, y rangesync.KeyBytes) error
	// Serve serves a synchronization request on the specified stream.
	// It returns a sequence of new keys that were received from the peer and the
	// number of received items.
	Serve(ctx context.Context, p p2p.Peer, stream io.ReadWriter) error
	// Probe probes the specified peer, obtaining its set fingerprint,
	// the number of items and the similarity value.
	Probe(ctx context.Context, p p2p.Peer) (rangesync.ProbeResult, error)
}

SyncBase is a synchronization base which holds the original OrderedSet. It is used to sync against peers using derived OrderedSets. It can also probe peers to decide on the synchronization strategy.

type SyncKeyHandler

type SyncKeyHandler interface {
	// Commit is invoked at the end of synchronization to apply the changes.
	Commit(ctx context.Context, peer p2p.Peer, base rangesync.OrderedSet, received rangesync.SeqResult) error
}

SyncKeyHandler is a handler for keys that are received from peers.