From 8624668e6cf7508d94ae513451cc6a6badf98e2c Mon Sep 17 00:00:00 2001 From: Javier Peletier Date: Tue, 2 Apr 2019 23:50:17 +0200 Subject: [PATCH] swarm/storage/feeds/lookup: Commented tests and LongEarthAlgorithm --- .../feed/lookup/algorithm_longearth.go | 72 +++++++++++++------ swarm/storage/feed/lookup/lookup_test.go | 4 +- swarm/storage/feed/lookup/store_test.go | 52 ++++++++++---- 3 files changed, 92 insertions(+), 36 deletions(-) diff --git a/swarm/storage/feed/lookup/algorithm_longearth.go b/swarm/storage/feed/lookup/algorithm_longearth.go index 77ce7d4f16..152d00da9f 100644 --- a/swarm/storage/feed/lookup/algorithm_longearth.go +++ b/swarm/storage/feed/lookup/algorithm_longearth.go @@ -6,43 +6,66 @@ import ( "time" ) -type StepFunc func(ctx context.Context, t uint64, hint Epoch) interface{} +type stepFunc func(ctx context.Context, t uint64, hint Epoch) interface{} +// LongEarthLookaheadDelay is the headstart the lookahead gives R before it launches +var LongEarthLookaheadDelay = 250 * time.Millisecond + +// LongEarthLookbackDelay is the headstart the lookback gives R before it launches +var LongEarthLookbackDelay = 250 * time.Millisecond + +// LongEarthAlgorithm explores possible lookup paths in parallel, pruning paths as soon +// as a more promising lookup path is found. As a result, this lookup algorithm is an order +// of magnitude faster than the FluzCapacitor algorithm, but at the expense of more exploratory reads. +// This algorithm works as follows. On each step, the next epoch is immediately looked up (R) +// and given a head start, while two parallel "steps" are launched a short time after: +// look ahead (A) is the path the algorithm would take if the R lookup returns a value, whereas +// look back (B) is the path the algorithm would take if the R lookup failed. +// as soon as R is actually finished, the A or B paths are pruned depending on the value of R. +// if A returns earlier than R, then R and B read operations can be safely canceled, saving time. +// The maximum number of active read operations is calculated as 2^(timeout/headstart). +// If headstart is infinite, this algorithm behaves as FluzCapacitor. +// timeout is the maximum execution time of the passed `read` function. +// the two head starts can be configured by changing LongEarthLookaheadDelay or LongEarthLookbackDelay func LongEarthAlgorithm(ctx context.Context, now uint64, hint Epoch, read ReadFunc) (interface{}, error) { - var stepCounter int32 + var stepCounter int32 // for debugging, stepCounter allows to give an ID to each step instance - errc := make(chan struct{}) - var gerr error + errc := make(chan struct{}) // errc will help as an error shortcut signal + var gerr error // in case of error, this variable will be set - var step StepFunc + var step stepFunc // For efficiency, the algorithm step is defined as a closure step = func(ctxS context.Context, t uint64, hint Epoch) interface{} { - stepID := atomic.AddInt32(&stepCounter, 1) + stepID := atomic.AddInt32(&stepCounter, 1) // give an ID to this call instance trace(stepID, "init: t=%d, hint=%s", t, hint.String()) var valueA, valueB, valueR interface{} - ctxR, cancelR := context.WithCancel(ctxS) - ctxA, cancelA := context.WithCancel(ctxS) - ctxB, cancelB := context.WithCancel(ctxS) + // initialize the three read contexts + ctxR, cancelR := context.WithCancel(ctxS) // will handle the current read operation + ctxA, cancelA := context.WithCancel(ctxS) // will handle the lookahead path + ctxB, cancelB := context.WithCancel(ctxS) // will handle the lookback path - epoch := GetNextEpoch(hint, t) + epoch := GetNextEpoch(hint, t) // calculate the epoch to look up in this step instance + // define the lookAhead function, which will follow the path as if R was successful lookAhead := func() { - valueA = step(ctxA, t, epoch) - if valueA != nil { + valueA = step(ctxA, t, epoch) // launch the next step, recursively. + if valueA != nil { // if this path is successful, we don't need R or B. cancelB() cancelR() } } + // define the lookBack function, which will follow the path as if R was unsuccessful lookBack := func() { - var err error if epoch.Base() == hint.Base() { // we have reached the hint itself if hint == worstHint { valueB = nil return } - // check it out + var err error + // check the hint. If successful, this is it. Otherwise + // we've been given a bad hint! valueB, err = read(ctxB, hint, now) if valueB != nil || err == context.Canceled { return @@ -63,11 +86,11 @@ func LongEarthAlgorithm(ctx context.Context, now uint64, hint Epoch, read ReadFu valueB = step(ctxB, base-1, hint) } - go func() { + go func() { //goroutine to read the current epoch (R) defer cancelR() var err error - valueR, err = read(ctxR, epoch, now) - if valueR == nil { + valueR, err = read(ctxR, epoch, now) // read this epoch + if valueR == nil { // if unsuccessful, cancel lookahead, otherwise cancel lookback. cancelA() } else { cancelB() @@ -78,15 +101,17 @@ func LongEarthAlgorithm(ctx context.Context, now uint64, hint Epoch, read ReadFu } }() - go func() { + go func() { // goroutine to give a headstart to R and then launch lookahead. defer cancelA() + // if we are at the lowest level or this is the hint, then we cannot lookahead if epoch.Level == LowestLevel || epoch.Equals(hint) { return } + // give a head start to R, or launch immediately if R finishes early enough select { - case <-TimeAfter(250 * time.Millisecond): + case <-TimeAfter(LongEarthLookaheadDelay): lookAhead() case <-ctxR.Done(): if valueR != nil { @@ -96,11 +121,12 @@ func LongEarthAlgorithm(ctx context.Context, now uint64, hint Epoch, read ReadFu } }() - go func() { + go func() { // goroutine to give a headstart to R and then launch lookback. defer cancelB() + // give a head start to R, or launch immediately if R finishes early enough select { - case <-TimeAfter(250 * time.Millisecond): + case <-TimeAfter(LongEarthLookbackDelay): lookBack() case <-ctxR.Done(): if valueR == nil { @@ -130,11 +156,13 @@ func LongEarthAlgorithm(ctx context.Context, now uint64, hint Epoch, read ReadFu stepCtx, cancel := context.WithCancel(ctx) defer cancel() - go func() { + go func() { // launch the root step in its own goroutine to allow cancellation value = step(stepCtx, now, hint) cancel() }() + // wait for the algorithm to finish, but shortcut in case + // of errors select { case <-stepCtx.Done(): return value, ctx.Err() diff --git a/swarm/storage/feed/lookup/lookup_test.go b/swarm/storage/feed/lookup/lookup_test.go index 3458485e72..6a55683157 100644 --- a/swarm/storage/feed/lookup/lookup_test.go +++ b/swarm/storage/feed/lookup/lookup_test.go @@ -40,8 +40,8 @@ const enablePrintMetrics = true // set to true to display algorithm benchmarking func printMetric(metric string, store *Store, elapsed time.Duration) { if enablePrintMetrics { - fmt.Printf("metric=%s, readcount=%d (successful=%d, failed=%d), cached=%d, canceled=%d, elapsed=%s\n", metric, - store.reads, store.sucessful, store.failed, store.cacheHits, store.canceled, elapsed) + fmt.Printf("metric=%s, readcount=%d (successful=%d, failed=%d), cached=%d, canceled=%d, maxSimult=%d, elapsed=%s\n", metric, + store.reads, store.sucessful, store.failed, store.cacheHits, store.canceled, store.maxSimultaneous, elapsed) } } diff --git a/swarm/storage/feed/lookup/store_test.go b/swarm/storage/feed/lookup/store_test.go index 8d349015e5..148034687a 100644 --- a/swarm/storage/feed/lookup/store_test.go +++ b/swarm/storage/feed/lookup/store_test.go @@ -1,5 +1,10 @@ package lookup_test +/* +This file contains components to mock a storage for testing +lookup algorithms and measure the number of reads. +*/ + import ( "context" "fmt" @@ -10,39 +15,49 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage/feed/lookup" ) +// Data is a struct to keep a value to store/retrieve during testing type Data struct { Payload uint64 Time uint64 } +// String implements fmt.Stringer func (d *Data) String() string { return fmt.Sprintf("%d-%d", d.Payload, d.Time) } +// Datamap is an internal map to hold the mocked storage type DataMap map[lookup.EpochID]*Data +// StoreConfig allows to specify the simulated delays for each type of +// read operation type StoreConfig struct { - CacheReadTime time.Duration - FailedReadTime time.Duration - SuccessfulReadTime time.Duration + CacheReadTime time.Duration // time it takes to read from the cache + FailedReadTime time.Duration // time it takes to acknowledge a read as failed + SuccessfulReadTime time.Duration // time it takes to fetch data } +// StoreCounters will track read count metrics type StoreCounters struct { - reads int - cacheHits int - failed int - sucessful int - canceled int + reads int + cacheHits int + failed int + sucessful int + canceled int + maxSimultaneous int } +// Store simulates a store and keeps track of performance counters type Store struct { StoreConfig StoreCounters - data DataMap - cache DataMap - lock sync.RWMutex + data DataMap + cache DataMap + lock sync.RWMutex + activeReads int } +// NewStore returns a new mock store ready for use func NewStore(config *StoreConfig) *Store { store := &Store{ StoreConfig: *config, @@ -53,26 +68,29 @@ func NewStore(config *StoreConfig) *Store { return store } +// Reset reset performance counters and clears the cache func (s *Store) Reset() { s.cache = make(DataMap) s.StoreCounters = StoreCounters{} } +// Put stores a value in the mock store at the given epoch func (s *Store) Put(epoch lookup.Epoch, value *Data) { log.Debug("Write: %d-%d, value='%d'\n", epoch.Base(), epoch.Level, value.Payload) s.data[epoch.ID()] = value } +// Update runs the seed algorithm to place the update in the appropriate epoch func (s *Store) Update(last lookup.Epoch, now uint64, value *Data) lookup.Epoch { epoch := lookup.GetNextEpoch(last, now) s.Put(epoch, value) return epoch } +// Get retrieves data at the specified epoch, simulating a delay func (s *Store) Get(ctx context.Context, epoch lookup.Epoch, now uint64) (value interface{}, err error) { epochID := epoch.ID() var operationTime time.Duration - s.reads++ defer func() { // simulate a delay according to what has actually happened select { @@ -84,10 +102,18 @@ func (s *Store) Get(ctx context.Context, epoch lookup.Epoch, now uint64) (value value = nil err = ctx.Err() } + s.lock.Lock() + s.activeReads-- + s.lock.Unlock() }() s.lock.Lock() defer s.lock.Unlock() + s.reads++ + s.activeReads++ + if s.activeReads > s.maxSimultaneous { + s.maxSimultaneous = s.activeReads + } // 1.- Simulate a cache read item := s.cache[epochID] @@ -119,6 +145,8 @@ func (s *Store) Get(ctx context.Context, epoch lookup.Epoch, now uint64) (value return nil, nil } +// MakeReadFunc returns a read function suitable for the lookup algorithm, mapped +// to this mock storage func (s *Store) MakeReadFunc() lookup.ReadFunc { return func(ctx context.Context, epoch lookup.Epoch, now uint64) (interface{}, error) { return s.Get(ctx, epoch, now)