diff --git a/pstoreds/addr_book.go b/pstoreds/addr_book.go index cf766ba..624f35e 100644 --- a/pstoreds/addr_book.go +++ b/pstoreds/addr_book.go @@ -31,7 +31,7 @@ const ( var ( log = logging.Logger("peerstore/ds") - // Peer addresses are stored under the following db key pattern: + // Peer addresses are stored db key pattern: // /peers/addrs/ addrBookBase = ds.NewKey("/peers/addrs") ) @@ -44,7 +44,7 @@ type addrsRecord struct { } // flush writes the record to the datastore by calling ds.Put, unless the record is -// marked for deletion, in which case we call ds.Delete. +// marked for deletion, in which case we call ds.Delete. To be called within a lock. func (r *addrsRecord) flush(write ds.Write) (err error) { key := addrBookBase.ChildString(b32.RawStdEncoding.EncodeToString([]byte(r.Id.ID))) if len(r.Addrs) == 0 { @@ -63,20 +63,19 @@ func (r *addrsRecord) flush(write ds.Write) (err error) { return nil } -// Clean is called on records to perform housekeeping. The return value signals if the record was changed -// as a result of the cleaning. +// Clean is called on records to perform housekeeping. The return value indicates if the record was changed +// as a result of this call. // // Clean does the following: -// * sorts the addresses by expiration (soonest expiring first). -// * removes the addresses that have expired. +// * sorts addresses by expiration (soonest expiring first). +// * removes expired addresses. // -// It short-circuits optimistically when we know there's nothing to do. +// It short-circuits optimistically when there's nothing to do. // // Clean is called from several points: -// * when accessing and loading an entry. +// * when accessing an entry. // * when performing periodic GC. -// * after an entry has been modified (e.g. addresses have been added or removed, -// TTLs updated, etc.) +// * after an entry has been modified (e.g. addresses have been added or removed, TTLs updated, etc.) // // If the return value is true, the caller can perform a flush immediately, or can schedule an async // flush, depending on the context. @@ -101,8 +100,8 @@ func (r *addrsRecord) Clean() (chgd bool) { }) } - // since addresses are sorted by expiration, we find the first survivor and split the - // slice on its index. + // since addresses are sorted by expiration, we find the first + // survivor and split the slice on its index. pivot := -1 for i, addr := range r.Addrs { if addr.Expiry > now { @@ -115,8 +114,8 @@ func (r *addrsRecord) Clean() (chgd bool) { return r.dirty || pivot >= 0 } -// dsAddrBook is an address book backed by a Datastore with a GC-like procedure -// to purge expired entries. It uses an in-memory address stream manager. +// dsAddrBook is an address book backed by a Datastore with a GC procedure to purge expired entries. It uses an +// in-memory address stream manager. See the NewAddrBook for more information. type dsAddrBook struct { ctx context.Context opts Options @@ -133,8 +132,25 @@ type dsAddrBook struct { var _ pstore.AddrBook = (*dsAddrBook)(nil) -// NewAddrBook initializes a new address book given a Datastore instance, a context for managing the TTL manager, -// and the interval at which the TTL manager should sweep the Datastore. +// NewAddrBook initializes a new datastore-backed address book. It serves as a drop-in replacement for pstoremem +// (memory-backed peerstore), and works with any datastore implementing the ds.Batching interface. +// +// Addresses and peer records are serialized into protobuf, storing one datastore entry per peer, along with metadata +// to control address expiration. To alleviate disk access and serde overhead, we internally use a read/write-through +// ARC cache, the size of which is adjustable via Options.CacheSize. +// +// The user has a choice of two GC algorithms: +// +// - lookahead GC: minimises the amount of full store traversals by maintaining a time-indexed list of entries that +// need to be visited within the period specified in Options.GCLookaheadInterval. This is useful in scenarios with +// considerable TTL variance, coupled with datastores whose native iterators return entries in lexicographical key +// order. Enable this mode by passing a value Options.GCLookaheadInterval > 0. Lookahead windows are jumpy, not +// sliding. Purges operate exclusively over the lookahead window with periodicity Options.GCPurgeInterval. +// +// - full-purge GC (default): performs a full visit of the store with periodicity Options.GCPurgeInterval. Useful when +// the range of possible TTL values is small and the values themselves are also extreme, e.g. 10 minutes or +// permanent, popular values used in other libp2p modules. In this cited case, optimizing with lookahead windows +// makes little sense. func NewAddrBook(ctx context.Context, store ds.Batching, opts Options) (ab *dsAddrBook, err error) { var cache cache = new(noopCache) if opts.CacheSize > 0 { @@ -158,7 +174,6 @@ func NewAddrBook(ctx context.Context, store ds.Batching, opts Options) (ab *dsAd return nil, err } - // kick off background processes. go ab.flusher() return ab, nil @@ -180,7 +195,7 @@ func (ab *dsAddrBook) asyncFlush(pr *addrsRecord) { // loadRecord is a read-through fetch. It fetches a record from cache, falling back to the // datastore upon a miss, and returning a newly initialized record if the peer doesn't exist. // -// loadRecord calls Clean() on the record before returning it. If the record changes +// loadRecord calls Clean() on existing recordsrecord before returning it. If the record changes // as a result and the update argument is true, an async flush is queued. // // If the cache argument is true, the record is inserted in the cache when loaded from the datastore. diff --git a/pstoreds/addr_book_gc.go b/pstoreds/addr_book_gc.go index 931dad2..61444ab 100644 --- a/pstoreds/addr_book_gc.go +++ b/pstoreds/addr_book_gc.go @@ -16,13 +16,12 @@ import ( ) var ( - // GC lookahead entries are stored in keys with pattern: + // GC lookahead entries are stored in key pattern: // /peers/gc/addrs// => nil + // in databases with lexicographical key order, this time-indexing allows us to visit + // only the timeslice we are interested in. gcLookaheadBase = ds.NewKey("/peers/gc/addrs") - // in GC routines, how many operations do we place in a batch before it's committed. - gcOpsPerBatch = 20 - // queries purgeLookaheadQuery = query.Query{ Prefix: gcLookaheadBase.String(), @@ -43,7 +42,6 @@ var ( } ) -// dsAddrBookGc encapsulates the GC behaviour to maintain a datastore-backed address book. type dsAddrBookGc struct { ctx context.Context ab *dsAddrBook @@ -63,6 +61,10 @@ func newAddressBookGc(ctx context.Context, ab *dsAddrBook) (*dsAddrBookGc, error if ab.opts.GCInitialDelay < 0 { return nil, fmt.Errorf("negative GC initial delay provided: %s", ab.opts.GCInitialDelay) } + if ab.opts.GCLookaheadInterval > 0 && ab.opts.GCLookaheadInterval < ab.opts.GCPurgeInterval { + return nil, fmt.Errorf("lookahead interval must be larger than purge interval, respectively: %s, %s", + ab.opts.GCLookaheadInterval, ab.opts.GCPurgeInterval) + } lookaheadEnabled := ab.opts.GCLookaheadInterval > 0 gc := &dsAddrBookGc{ @@ -136,7 +138,7 @@ func (gc *dsAddrBookGc) purgeLookahead() { var id peer.ID record := &addrsRecord{AddrBookRecord: &pb.AddrBookRecord{}} // empty record to reuse and avoid allocs. - batch, err := newCyclicBatch(gc.ab.ds) + batch, err := newCyclicBatch(gc.ab.ds, defaultOpsPerCyclicBatch) if err != nil { log.Warningf("failed while creating batch to purge GC entries: %v", err) } @@ -259,7 +261,7 @@ func (gc *dsAddrBookGc) purgeStore() { } record := &addrsRecord{AddrBookRecord: &pb.AddrBookRecord{}} // empty record to reuse and avoid allocs. - batch, err := newCyclicBatch(gc.ab.ds) + batch, err := newCyclicBatch(gc.ab.ds, defaultOpsPerCyclicBatch) if err != nil { log.Warningf("failed while creating batch to purge GC entries: %v", err) } @@ -324,7 +326,7 @@ func (gc *dsAddrBookGc) populateLookahead() { } defer results.Close() - batch, err := newCyclicBatch(gc.ab.ds) + batch, err := newCyclicBatch(gc.ab.ds, defaultOpsPerCyclicBatch) if err != nil { log.Warningf("failed while creating batch to populate lookahead GC window: %v", err) return diff --git a/pstoreds/cyclic_batch.go b/pstoreds/cyclic_batch.go index 64dc103..1cf8579 100644 --- a/pstoreds/cyclic_batch.go +++ b/pstoreds/cyclic_batch.go @@ -6,18 +6,22 @@ import ( ds "github.com/ipfs/go-datastore" ) -// cyclicBatch buffers ds write operations and automatically flushes them after gcOpsPerBatch (20) have been +// how many operations are queued in a cyclic batch before we flush it. +var defaultOpsPerCyclicBatch = 20 + +// cyclicBatch buffers ds write operations and automatically flushes them after defaultOpsPerCyclicBatch (20) have been // queued. An explicit `Commit()` closes this cyclic batch, erroring all further operations. // // It is similar to go-ds autobatch, but it's driven by an actual Batch facility offered by the // ds. type cyclicBatch struct { + threshold int ds.Batch ds ds.Batching pending int } -func newCyclicBatch(ds ds.Batching) (ds.Batch, error) { +func newCyclicBatch(ds ds.Batching, threshold int) (ds.Batch, error) { batch, err := ds.Batch() if err != nil { return nil, err @@ -29,7 +33,7 @@ func (cb *cyclicBatch) cycle() (err error) { if cb.Batch == nil { return errors.New("cyclic batch is closed") } - if cb.pending < gcOpsPerBatch { + if cb.pending < cb.threshold { // we haven't reached the threshold yet. return nil }