diff --git a/swarm/storage/feed/lookup/algorithm_fluzcapacitor.go b/swarm/storage/feed/lookup/algorithm_fluzcapacitor.go new file mode 100644 index 0000000000..3840bd0fd6 --- /dev/null +++ b/swarm/storage/feed/lookup/algorithm_fluzcapacitor.go @@ -0,0 +1,63 @@ +package lookup + +import "context" + +// FluzCapacitorAlgorithm works by narrowing the epoch search area if an update is found +// going back and forth in time +// First, it will attempt to find an update where it should be now if the hint was +// really the last update. If that lookup fails, then the last update must be either the hint itself +// or the epochs right below. If however, that lookup succeeds, then the update must be +// that one or within the epochs right below. +// see the guide for a more graphical representation +func FluzCapacitorAlgorithm(ctx context.Context, now uint64, hint Epoch, read ReadFunc) (value interface{}, err error) { + var lastFound interface{} + var epoch Epoch + if hint == NoClue { + hint = worstHint + } + + t := now + + for { + epoch = GetNextEpoch(hint, t) + value, err = read(ctx, epoch, now) + if err != nil { + return nil, err + } + if value != nil { + lastFound = value + if epoch.Level == LowestLevel || epoch.Equals(hint) { + return value, nil + } + hint = epoch + continue + } + if epoch.Base() == hint.Base() { + if lastFound != nil { + return lastFound, nil + } + // we have reached the hint itself + if hint == worstHint { + return nil, nil + } + // check it out + value, err = read(ctx, hint, now) + if err != nil { + return nil, err + } + if value != nil { + return value, nil + } + // bad hint. + t = hint.Base() + hint = worstHint + continue + } + base := epoch.Base() + if base == 0 { + return nil, nil + } + t = base - 1 + } + +} diff --git a/swarm/storage/feed/lookup/algorithm_longearth.go b/swarm/storage/feed/lookup/algorithm_longearth.go new file mode 100644 index 0000000000..77ce7d4f16 --- /dev/null +++ b/swarm/storage/feed/lookup/algorithm_longearth.go @@ -0,0 +1,144 @@ +package lookup + +import ( + "context" + "sync/atomic" + "time" +) + +type StepFunc func(ctx context.Context, t uint64, hint Epoch) interface{} + +func LongEarthAlgorithm(ctx context.Context, now uint64, hint Epoch, read ReadFunc) (interface{}, error) { + var stepCounter int32 + + errc := make(chan struct{}) + var gerr error + + var step StepFunc + step = func(ctxS context.Context, t uint64, hint Epoch) interface{} { + stepID := atomic.AddInt32(&stepCounter, 1) + 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) + + epoch := GetNextEpoch(hint, t) + + lookAhead := func() { + valueA = step(ctxA, t, epoch) + if valueA != nil { + cancelB() + cancelR() + } + } + + 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 + valueB, err = read(ctxB, hint, now) + if valueB != nil || err == context.Canceled { + return + } + if err != nil { + gerr = err + close(errc) + return + } + // bad hint. + valueB = step(ctxB, hint.Base(), worstHint) + return + } + base := epoch.Base() + if base == 0 { + return + } + valueB = step(ctxB, base-1, hint) + } + + go func() { + defer cancelR() + var err error + valueR, err = read(ctxR, epoch, now) + if valueR == nil { + cancelA() + } else { + cancelB() + } + if err != nil && err != context.Canceled { + gerr = err + close(errc) + } + }() + + go func() { + defer cancelA() + + if epoch.Level == LowestLevel || epoch.Equals(hint) { + return + } + + select { + case <-TimeAfter(250 * time.Millisecond): + lookAhead() + case <-ctxR.Done(): + if valueR != nil { + lookAhead() + } + case <-ctxA.Done(): + } + }() + + go func() { + defer cancelB() + + select { + case <-TimeAfter(250 * time.Millisecond): + lookBack() + case <-ctxR.Done(): + if valueR == nil { + lookBack() + } + case <-ctxB.Done(): + } + }() + + <-ctxA.Done() + if valueA != nil { + trace(stepID, "Returning valueA=%v", valueA) + return valueA + } + + <-ctxR.Done() + if valueR != nil { + trace(stepID, "Returning valueR=%v", valueR) + return valueR + } + <-ctxB.Done() + trace(stepID, "Returning valueB=%v", valueB) + return valueB + } + + var value interface{} + stepCtx, cancel := context.WithCancel(ctx) + defer cancel() + + go func() { + value = step(stepCtx, now, hint) + cancel() + }() + + select { + case <-stepCtx.Done(): + return value, ctx.Err() + case <-errc: + return nil, gerr + } +} diff --git a/swarm/storage/feed/lookup/lookup.go b/swarm/storage/feed/lookup/lookup.go index e6f13d675e..3f9f29cbbb 100644 --- a/swarm/storage/feed/lookup/lookup.go +++ b/swarm/storage/feed/lookup/lookup.go @@ -48,6 +48,11 @@ type Algorithm func(ctx context.Context, now uint64, hint Epoch, read ReadFunc) // Returns nil if an update was not found var Lookup Algorithm = LongEarthAlgorithm +// TimeAfter must point to a function that returns a timer +// This is here so that tests can replace it with +// a mock up timer factory to simulate time deterministically +var TimeAfter = time.After + // ReadFunc is a handler called by Lookup each time it attempts to find a value // It should return if a value is not found // It should return if a value is found, but its timestamp is higher than "now" @@ -126,190 +131,6 @@ func GetFirstEpoch(now uint64) Epoch { var worstHint = Epoch{Time: 0, Level: 63} -// FluzCapacitorAlgorithm works by narrowing the epoch search area if an update is found -// going back and forth in time -// First, it will attempt to find an update where it should be now if the hint was -// really the last update. If that lookup fails, then the last update must be either the hint itself -// or the epochs right below. If however, that lookup succeeds, then the update must be -// that one or within the epochs right below. -// see the guide for a more graphical representation -func FluzCapacitorAlgorithm(ctx context.Context, now uint64, hint Epoch, read ReadFunc) (value interface{}, err error) { - var lastFound interface{} - var epoch Epoch - if hint == NoClue { - hint = worstHint - } - - t := now - - for { - epoch = GetNextEpoch(hint, t) - value, err = read(ctx, epoch, now) - if err != nil { - return nil, err - } - if value != nil { - lastFound = value - if epoch.Level == LowestLevel || epoch.Equals(hint) { - return value, nil - } - hint = epoch - continue - } - if epoch.Base() == hint.Base() { - if lastFound != nil { - return lastFound, nil - } - // we have reached the hint itself - if hint == worstHint { - return nil, nil - } - // check it out - value, err = read(ctx, hint, now) - if err != nil { - return nil, err - } - if value != nil { - return value, nil - } - // bad hint. - t = hint.Base() - hint = worstHint - continue - } - base := epoch.Base() - if base == 0 { - return nil, nil - } - t = base - 1 - } - -} - -type StepFunc func(ctx context.Context, t uint64, hint Epoch) interface{} - -func LongEarthAlgorithm(ctx context.Context, now uint64, hint Epoch, read ReadFunc) (interface{}, error) { - - errc := make(chan error) - - var step StepFunc - step = func(ctxS context.Context, t uint64, hint Epoch) interface{} { - var valueA, valueB, valueR interface{} - - ctxR, cancelR := context.WithCancel(ctxS) - ctxA, cancelA := context.WithCancel(ctxS) - ctxB, cancelB := context.WithCancel(ctxS) - - epoch := GetNextEpoch(hint, t) - - lookAhead := func() { - valueA = step(ctxA, t, epoch) - if valueA != nil { - cancelB() - cancelR() - } - } - - 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 - valueB, err = read(ctxB, hint, now) - if valueB != nil || err == context.Canceled { - return - } - if err != nil { - errc <- err - return - } - // bad hint. - valueB = step(ctxB, hint.Base(), worstHint) - return - } - base := epoch.Base() - if base == 0 { - return - } - valueB = step(ctxB, base-1, hint) - } - - go func() { - defer cancelR() - var err error - valueR, err = read(ctxR, epoch, now) - if valueR == nil { - cancelA() - } else { - cancelB() - } - if err != nil && err != context.Canceled { - errc <- err - } - }() - - go func() { - defer cancelA() - - if epoch.Level == LowestLevel || epoch.Equals(hint) { - return - } - - select { - case <-time.After(250 * time.Millisecond): - lookAhead() - case <-ctxR.Done(): - if valueR != nil { - lookAhead() - } - case <-ctxA.Done(): - } - }() - - go func() { - defer cancelB() - - select { - case <-time.After(250 * time.Millisecond): - lookBack() - case <-ctxR.Done(): - if valueR == nil { - lookBack() - } - case <-ctxB.Done(): - } - }() - - <-ctxA.Done() - if valueA != nil { - return valueA - } - - <-ctxR.Done() - if valueR != nil { - return valueR - } - <-ctxB.Done() - return valueB - } - - var value interface{} - ctx, cancel := context.WithCancel(ctx) - defer cancel() - - go func() { - value = step(ctx, now, hint) - cancel() - }() - - select { - case <-ctx.Done(): - return value, nil - case err := <-errc: - return nil, err - } +var trace = func(id int32, formatString string, a ...interface{}) { + //fmt.Printf("Step ID #%d "+formatString+"\n", append([]interface{}{id}, a...)...) } diff --git a/swarm/storage/feed/lookup/lookup_test.go b/swarm/storage/feed/lookup/lookup_test.go index 3322d2f017..ac0d1ddb84 100644 --- a/swarm/storage/feed/lookup/lookup_test.go +++ b/swarm/storage/feed/lookup/lookup_test.go @@ -28,6 +28,25 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage/feed/lookup" ) +type AlgorithmInfo struct { + Lookup lookup.Algorithm + Name string +} + +var algorithms = []AlgorithmInfo{ + {lookup.FluzCapacitorAlgorithm, "FluzCapacitor"}, + {lookup.LongEarthAlgorithm, "LongEarth"}, +} + +const enablePrintMetrics = true // set to true to display algorithm benchmarking stats + +func printMetric(metric string, reads int32, elapsed time.Duration) { + if !enablePrintMetrics { + return + } + fmt.Printf("metric=%s, readcount=%d, elapsed=%s\n", metric, reads, elapsed) +} + type Data struct { Payload uint64 Time uint64 @@ -60,7 +79,7 @@ func makeReadFunc(store Store, counter *int32) lookup.ReadFunc { return func(ctx context.Context, epoch lookup.Epoch, now uint64) (interface{}, error) { atomic.AddInt32(counter, 1) select { - case <-time.After(1000 * time.Millisecond): + case <-lookup.TimeAfter(1000 * time.Millisecond): case <-ctx.Done(): return nil, ctx.Err() } @@ -70,7 +89,7 @@ func makeReadFunc(store Store, counter *int32) lookup.ReadFunc { valueStr = fmt.Sprintf("%d", data.Payload) } log.Debug("Read: %d-%d, value='%s'\n", epoch.Base(), epoch.Level, valueStr) - //fmt.Printf("Read: %d-%d, value='%s'\n", epoch.Base(), epoch.Level, valueStr) + if data != nil && data.Time <= now { return data, nil } @@ -79,6 +98,10 @@ func makeReadFunc(store Store, counter *int32) lookup.ReadFunc { } func TestLookup(t *testing.T) { + stopwatch := NewStopwatch(50 * time.Millisecond) + lookup.TimeAfter = stopwatch.TimeAfter() + defer stopwatch.Stop() + store := make(Store) var readCount int32 = 0 readFunc := makeReadFunc(store, &readCount) @@ -98,60 +121,78 @@ func TestLookup(t *testing.T) { lastData = &data } - // try to get the last value + for _, algo := range algorithms { + t.Run(algo.Name, func(t *testing.T) { + readCount = 0 + stopwatch.Reset() + stopwatch.Run() - value, err := lookup.Lookup(context.Background(), now, lookup.NoClue, readFunc) - if err != nil { - t.Fatal(err) + // try to get the last value + value, err := algo.Lookup(context.Background(), now, lookup.NoClue, readFunc) + stopwatch.Stop() + if err != nil { + t.Fatal(err) + } + timeElapsedWithoutHint := stopwatch.Elapsed() + printMetric("SIMPLE READ", readCount, timeElapsedWithoutHint) + + if value != lastData { + t.Fatalf("Expected lookup to return the last written value: %v. Got %v", lastData, value) + } + + // reset the read count for the next test + readCount = 0 + stopwatch.Reset() + stopwatch.Run() + // Provide a hint to get a faster lookup. In particular, we give the exact location of the last update + value, err = algo.Lookup(context.Background(), now, epoch, readFunc) + stopwatch.Stop() + if err != nil { + t.Fatal(err) + } + printMetric("WITH HINT", readCount, stopwatch.Elapsed()) + + if value != lastData { + t.Fatalf("Expected lookup to return the last written value: %v. Got %v", lastData, value) + } + + if stopwatch.Elapsed() > timeElapsedWithoutHint { + t.Fatalf("Expected lookup to complete faster than %s since we provided a hint. Took %s", timeElapsedWithoutHint, stopwatch.Elapsed()) + } + + // try to get an intermediate value + // if we look for a value in now - Year*3 + 6*Month, we should get that value + // Since the "payload" is the timestamp itself, we can check this. + + expectedTime := now - Year*3 + 6*Month + // reset the read count for the next test + readCount = 0 + stopwatch.Reset() + stopwatch.Run() + value, err = algo.Lookup(context.Background(), expectedTime, lookup.NoClue, readFunc) + stopwatch.Stop() + if err != nil { + t.Fatal(err) + } + printMetric("INTERMEDIATE READ", readCount, stopwatch.Elapsed()) + + data, ok := value.(*Data) + + if !ok { + t.Fatal("Expected value to contain data") + } + + if data.Time != expectedTime { + t.Fatalf("Expected value timestamp to be %d, got %d", data.Time, expectedTime) + } + }) } - fmt.Printf("readcount=%d\n", readCount) - - readCountWithoutHint := readCount - - if value != lastData { - t.Fatalf("Expected lookup to return the last written value: %v. Got %v", lastData, value) - } - - // reset the read count for the next test - readCount = 0 - // Provide a hint to get a faster lookup. In particular, we give the exact location of the last update - value, err = lookup.Lookup(context.Background(), now, epoch, readFunc) - if err != nil { - t.Fatal(err) - } - - if value != lastData { - t.Fatalf("Expected lookup to return the last written value: %v. Got %v", lastData, value) - } - - if readCount > readCountWithoutHint { - t.Fatalf("Expected lookup to complete with fewer or same reads than %d since we provided a hint. Did %d reads.", readCountWithoutHint, readCount) - } - - // try to get an intermediate value - // if we look for a value in now - Year*3 + 6*Month, we should get that value - // Since the "payload" is the timestamp itself, we can check this. - - expectedTime := now - Year*3 + 6*Month - - value, err = lookup.Lookup(context.Background(), expectedTime, lookup.NoClue, readFunc) - if err != nil { - t.Fatal(err) - } - - data, ok := value.(*Data) - - if !ok { - t.Fatal("Expected value to contain data") - } - - if data.Time != expectedTime { - t.Fatalf("Expected value timestamp to be %d, got %d", data.Time, expectedTime) - } - } func TestOneUpdateAt0(t *testing.T) { + stopwatch := NewStopwatch(50 * time.Millisecond) + lookup.TimeAfter = stopwatch.TimeAfter() + defer stopwatch.Stop() store := make(Store) var readCount int32 = 0 @@ -166,17 +207,29 @@ func TestOneUpdateAt0(t *testing.T) { } update(store, epoch, 0, &data) - value, err := lookup.Lookup(context.Background(), now, lookup.NoClue, readFunc) - if err != nil { - t.Fatal(err) - } - if value != &data { - t.Fatalf("Expected lookup to return the last written value: %v. Got %v", data, value) + for _, algo := range algorithms { + stopwatch.Reset() + t.Run(algo.Name, func(t *testing.T) { + readCount = 0 + stopwatch.Run() + value, err := algo.Lookup(context.Background(), now, lookup.NoClue, readFunc) + stopwatch.Stop() + if err != nil { + t.Fatal(err) + } + if value != &data { + t.Fatalf("Expected lookup to return the last written value: %v. Got %v", data, value) + } + printMetric("SIMPLE", readCount, stopwatch.Elapsed()) + }) } } // Tests the update is found even when a bad hint is given func TestBadHint(t *testing.T) { + stopwatch := NewStopwatch(50 * time.Millisecond) + lookup.TimeAfter = stopwatch.TimeAfter() + defer stopwatch.Stop() store := make(Store) var readCount int32 = 0 @@ -199,19 +252,32 @@ func TestBadHint(t *testing.T) { Time: 1200000000, } - value, err := lookup.Lookup(context.Background(), now, badHint, readFunc) - if err != nil { - t.Fatal(err) - } - if value != &data { - t.Fatalf("Expected lookup to return the last written value: %v. Got %v", data, value) + for _, algo := range algorithms { + stopwatch.Reset() + t.Run(algo.Name, func(t *testing.T) { + readCount = 0 + stopwatch.Run() + value, err := algo.Lookup(context.Background(), now, badHint, readFunc) + stopwatch.Stop() + if err != nil { + t.Fatal(err) + } + if value != &data { + t.Fatalf("Expected lookup to return the last written value: %v. Got %v", data, value) + } + printMetric("SIMPLE", readCount, stopwatch.Elapsed()) + }) } } // Tests whether the update is found when the bad hint is exactly below the last update func TestBadHintNextToUpdate(t *testing.T) { + stopwatch := NewStopwatch(50 * time.Millisecond) + lookup.TimeAfter = stopwatch.TimeAfter() + defer stopwatch.Stop() + store := make(Store) - readCount := 0 + var readCount int32 = 0 readFunc := makeReadFunc(store, &readCount) now := uint64(1533903729) @@ -248,13 +314,22 @@ func TestBadHintNextToUpdate(t *testing.T) { Level: 20, Time: 1200000005, } + for _, algo := range algorithms { + stopwatch.Reset() + t.Run(algo.Name, func(t *testing.T) { + readCount = 0 + stopwatch.Run() - value, err := lookup.Lookup(context.Background(), now, badHint, readFunc) - if err != nil { - t.Fatal(err) - } - if value != last { - t.Fatalf("Expected lookup to return the last written value: %v. Got %v", last, value) + value, err := algo.Lookup(context.Background(), now, badHint, readFunc) + stopwatch.Stop() + if err != nil { + t.Fatal(err) + } + if value != last { + t.Fatalf("Expected lookup to return the last written value: %v. Got %v", last, value) + } + printMetric("SIMPLE", readCount, stopwatch.Elapsed()) + }) } } @@ -265,50 +340,58 @@ func TestContextCancellation(t *testing.T) { return nil, ctx.Err() } - ctx, cancel := context.WithCancel(context.Background()) + for _, algo := range algorithms { + t.Run(algo.Name, func(t *testing.T) { - errc := make(chan error) + ctx, cancel := context.WithCancel(context.Background()) - go func() { - _, err := lookup.Lookup(ctx, 1200000000, lookup.NoClue, readFunc) - errc <- err - }() + errc := make(chan error) - cancel() + go func() { + _, err := lookup.Lookup(ctx, 1200000000, lookup.NoClue, readFunc) + errc <- err + }() - if err := <-errc; err != context.Canceled { - t.Fatalf("Expected lookup to return a context Cancelled error, got %v", err) - } + cancel() - // text context cancellation during hint lookup: - ctx, cancel = context.WithCancel(context.Background()) - errc = make(chan error) - someHint := lookup.Epoch{ - Level: 25, - Time: 300, - } + if err := <-errc; err != context.Canceled { + t.Fatalf("Expected lookup to return a context Cancelled error, got %v", err) + } - readFunc = func(ctx context.Context, epoch lookup.Epoch, now uint64) (interface{}, error) { - if epoch == someHint { - go cancel() - <-ctx.Done() - return nil, ctx.Err() - } - return nil, nil - } + // text context cancellation during hint lookup: + ctx, cancel = context.WithCancel(context.Background()) + errc = make(chan error) + someHint := lookup.Epoch{ + Level: 25, + Time: 300, + } - go func() { - _, err := lookup.Lookup(ctx, 301, someHint, readFunc) - errc <- err - }() + readFunc = func(ctx context.Context, epoch lookup.Epoch, now uint64) (interface{}, error) { + if epoch == someHint { + go cancel() + <-ctx.Done() + return nil, ctx.Err() + } + return nil, nil + } - if err := <-errc; err != context.Canceled { - t.Fatalf("Expected lookup to return a context Cancelled error, got %v", err) + go func() { + _, err := algo.Lookup(ctx, 301, someHint, readFunc) + errc <- err + }() + + if err := <-errc; err != context.Canceled { + t.Fatalf("Expected lookup to return a context Cancelled error, got %v", err) + } + }) } } func TestLookupFail(t *testing.T) { + stopwatch := NewStopwatch(50 * time.Millisecond) + lookup.TimeAfter = stopwatch.TimeAfter() + defer stopwatch.Stop() store := make(Store) var readCount int32 = 0 @@ -316,24 +399,33 @@ func TestLookupFail(t *testing.T) { readFunc := makeReadFunc(store, &readCount) now := uint64(1533903729) - // don't write anything and try to look up. - // we're testing we don't get stuck in a loop + for _, algo := range algorithms { + stopwatch.Reset() + t.Run(algo.Name, func(t *testing.T) { + readCount = 0 + stopwatch.Run() - value, err := lookup.Lookup(context.Background(), now, lookup.NoClue, readFunc) - if err != nil { - t.Fatal(err) - } - if value != nil { - t.Fatal("Expected value to be nil, since the update should've failed") - } + // don't write anything and try to look up. + // we're testing we don't get stuck in a loop - expectedReads := now/(1< readCountWithoutHint { - t.Fatalf("Expected lookup to complete with fewer or equal reads than %d since we provided a hint. Did %d reads.", readCountWithoutHint, readCount) - } + if value != lastData { + t.Fatalf("Expected lookup to return the last written value: %v. Got %v", lastData, value) + } - for i := uint64(0); i <= 994; i++ { - T := uint64(now - 1000 + i) // update every second for the last 1000 seconds - value, err := lookup.Lookup(context.Background(), T, lookup.NoClue, readFunc) - if err != nil { - t.Fatal(err) - } - data, _ := value.(*Data) - if data == nil { - t.Fatalf("Expected lookup to return %d, got nil", T) - } - if data.Payload != T { - t.Fatalf("Expected lookup to return %d, got %d", T, data.Time) - } + if stopwatch.Elapsed() > timeElapsedWithoutHint { + t.Fatalf("Expected lookup to complete faster than %s since we provided a hint. Took %s", timeElapsedWithoutHint, stopwatch.Elapsed()) + } + printMetric("WITH HINT", readCount, stopwatch.Elapsed()) + + readCount = 0 + stopwatch.Reset() + stopwatch.Run() + for i := uint64(0); i <= 10; i++ { + T := uint64(now - 1000 + i) // update every second for the last 1000 seconds + value, err := algo.Lookup(context.Background(), T, lookup.NoClue, readFunc) + if err != nil { + t.Fatal(err) + } + data, _ := value.(*Data) + if data == nil { + t.Fatalf("Expected lookup to return %d, got nil", T) + } + if data.Payload != T { + t.Fatalf("Expected lookup to return %d, got %d", T, data.Time) + } + } + stopwatch.Stop() + printMetric("MULTIPLE", readCount, stopwatch.Elapsed()) + }) } } func TestSparseUpdates(t *testing.T) { + stopwatch := NewStopwatch(50 * time.Millisecond) + lookup.TimeAfter = stopwatch.TimeAfter() + defer stopwatch.Stop() store := make(Store) var readCount int32 = 0 @@ -419,35 +533,49 @@ func TestSparseUpdates(t *testing.T) { lastData = &data } - // try to get the last value + for _, algo := range algorithms { + stopwatch.Reset() + t.Run(algo.Name, func(t *testing.T) { + readCount = 0 + stopwatch.Run() - value, err := lookup.Lookup(context.Background(), now, lookup.NoClue, readFunc) - if err != nil { - t.Fatal(err) + // try to get the last value + value, err := algo.Lookup(context.Background(), now, lookup.NoClue, readFunc) + stopwatch.Stop() + if err != nil { + t.Fatal(err) + } + + if value != lastData { + t.Fatalf("Expected lookup to return the last written value: %v. Got %v", lastData, value) + } + + timeElapsedWithoutHint := stopwatch.Elapsed() + printMetric("SIMPLE", readCount, timeElapsedWithoutHint) + + // reset the read count for the next test + readCount = 0 + stopwatch.Reset() + stopwatch.Run() + // Provide a hint to get a faster lookup. In particular, we give the exact location of the last update + value, err = algo.Lookup(context.Background(), now, epoch, readFunc) + stopwatch.Stop() + if err != nil { + t.Fatal(err) + } + + if value != lastData { + t.Fatalf("Expected lookup to return the last written value: %v. Got %v", lastData, value) + } + + if stopwatch.Elapsed() > timeElapsedWithoutHint { + t.Fatalf("Expected lookup to complete faster than %s since we provided a hint. Took %s", timeElapsedWithoutHint, stopwatch.Elapsed()) + } + + printMetric("WITH HINT", readCount, stopwatch.Elapsed()) + + }) } - - readCountWithoutHint := readCount - - if value != lastData { - t.Fatalf("Expected lookup to return the last written value: %v. Got %v", lastData, value) - } - - // reset the read count for the next test - readCount = 0 - // Provide a hint to get a faster lookup. In particular, we give the exact location of the last update - value, err = lookup.Lookup(context.Background(), now, epoch, readFunc) - if err != nil { - t.Fatal(err) - } - - if value != lastData { - t.Fatalf("Expected lookup to return the last written value: %v. Got %v", lastData, value) - } - - if readCount > readCountWithoutHint { - t.Fatalf("Expected lookup to complete with fewer reads than %d since we provided a hint. Did %d reads.", readCountWithoutHint, readCount) - } - } // testG will hold precooked test results @@ -525,14 +653,3 @@ func CookGetNextLevelTests(t *testing.T) { } fmt.Println(st) } - -func TestTest(t *testing.T) { - hint := lookup.Epoch{ - Time: 20, - Level: 2, - } - - e := lookup.GetNextEpoch(hint, 21) - - fmt.Println(e) -} diff --git a/swarm/storage/feed/lookup/timesim_test.go b/swarm/storage/feed/lookup/timesim_test.go new file mode 100644 index 0000000000..a269893eef --- /dev/null +++ b/swarm/storage/feed/lookup/timesim_test.go @@ -0,0 +1,94 @@ +package lookup_test + +import ( + "sync" + "sync/atomic" + "time" +) + +type Timer struct { + deadline time.Time + signal chan time.Time + id int32 +} + +type Stopwatch struct { + t time.Time + r time.Duration + timers map[int32]*Timer + timerCounter int32 + stopSignal chan struct{} + lock sync.RWMutex +} + +func NewStopwatch(resolution time.Duration) *Stopwatch { + s := &Stopwatch{ + r: resolution, + } + s.Reset() + return s +} + +func (s *Stopwatch) Reset() { + s.t = time.Time{} + s.timers = make(map[int32]*Timer) + s.Stop() +} + +func (s *Stopwatch) Tick() { + s.t = s.t.Add(s.r) + s.lock.Lock() + defer s.lock.Unlock() + for id, timer := range s.timers { + if s.t.After(timer.deadline) || s.t.Equal(timer.deadline) { + timer.signal <- s.t + close(timer.signal) + delete(s.timers, id) + } + } +} + +func (s *Stopwatch) GetTimer(duration time.Duration) <-chan time.Time { + timer := &Timer{ + deadline: s.t.Add(duration), + signal: make(chan time.Time, 1), + id: atomic.AddInt32(&s.timerCounter, 1), + } + s.lock.Lock() + defer s.lock.Unlock() + + s.timers[timer.id] = timer + return timer.signal +} + +func (s *Stopwatch) TimeAfter() func(d time.Duration) <-chan time.Time { + return func(d time.Duration) <-chan time.Time { + return s.GetTimer(d) + } +} + +func (s *Stopwatch) Elapsed() time.Duration { + return s.t.Sub(time.Time{}) +} + +func (s *Stopwatch) Run() { + go func() { + stopSignal := make(chan struct{}) + s.stopSignal = stopSignal + for { + select { + case <-time.After(1 * time.Millisecond): + s.Tick() + case <-stopSignal: + return + } + } + }() +} + +func (s *Stopwatch) Stop() { + if s.stopSignal != nil { + close(s.stopSignal) + s.stopSignal = nil + } +}