forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

swarm/storage/feed/lookup: multi-algorithm test suite

Browse source
This commit is contained in:
Javier Peletier 2019-04-01 18:45:04 +02:00
parent 9f31234beb
commit daf5126401
5 changed files with 610 additions and 371 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

63
swarm/storage/feed/lookup/algorithm_fluzcapacitor.go Normal file
View file

@ -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
}
}

144
swarm/storage/feed/lookup/algorithm_longearth.go Normal file
View file

@ -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
}
}

193
swarm/storage/feed/lookup/lookup.go
View file

@ -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 <nil> if a value is not found
// It should return <nil> 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...)...)
}

211
swarm/storage/feed/lookup/lookup_test.go
View file

@ -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,15 +121,20 @@ 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)
// try to get the last value
value, err := algo.Lookup(context.Background(), now, lookup.NoClue, readFunc)
stopwatch.Stop()
if err != nil {
t.Fatal(err)
}
fmt.Printf("readcount=%d\n", readCount)
readCountWithoutHint := readCount
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)
@ -114,18 +142,22 @@ func TestLookup(t *testing.T) {
// 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 = lookup.Lookup(context.Background(), now, epoch, readFunc)
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 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)
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
@ -133,11 +165,16 @@ func TestLookup(t *testing.T) {
// 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)
// 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)
@ -148,10 +185,14 @@ func TestLookup(t *testing.T) {
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)
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)
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,14 +314,23 @@ 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)
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())
})
}
}
func TestContextCancellation(t *testing.T) {
@ -265,6 +340,9 @@ func TestContextCancellation(t *testing.T) {
return nil, ctx.Err()
}
for _, algo := range algorithms {
t.Run(algo.Name, func(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
errc := make(chan error)
@ -298,17 +376,22 @@ func TestContextCancellation(t *testing.T) {
}
go func() {
_, err := lookup.Lookup(ctx, 301, someHint, readFunc)
_, 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,10 +399,17 @@ func TestLookupFail(t *testing.T) {
readFunc := makeReadFunc(store, &readCount)
now := uint64(1533903729)
for _, algo := range algorithms {
stopwatch.Reset()
t.Run(algo.Name, func(t *testing.T) {
readCount = 0
stopwatch.Run()
// don't write anything and try to look up.
// we're testing we don't get stuck in a loop
value, err := lookup.Lookup(context.Background(), now, lookup.NoClue, readFunc)
value, err := algo.Lookup(context.Background(), now, lookup.NoClue, readFunc)
stopwatch.Stop()
if err != nil {
t.Fatal(err)
}
@ -327,13 +417,15 @@ func TestLookupFail(t *testing.T) {
t.Fatal("Expected value to be nil, since the update should've failed")
}
expectedReads := now/(1<<lookup.HighestLevel) + 1
if uint64(readCount) != expectedReads {
t.Fatalf("Expected lookup to fail after %d reads. Did %d reads.", expectedReads, readCount)
printMetric("SIMPLE", readCount, stopwatch.Elapsed())
})
}
}
func TestHighFreqUpdates(t *testing.T) {
stopwatch := NewStopwatch(50 * time.Millisecond)
lookup.TimeAfter = stopwatch.TimeAfter()
defer stopwatch.Stop()
store := make(Store)
var readCount int32 = 0
@ -355,7 +447,14 @@ func TestHighFreqUpdates(t *testing.T) {
lastData = &data
}
value, err := lookup.Lookup(context.Background(), lastData.Time, lookup.NoClue, readFunc)
for _, algo := range algorithms {
stopwatch.Reset()
t.Run(algo.Name, func(t *testing.T) {
readCount = 0
stopwatch.Run()
value, err := algo.Lookup(context.Background(), lastData.Time, lookup.NoClue, readFunc)
stopwatch.Stop()
if err != nil {
t.Fatal(err)
}
@ -363,12 +462,16 @@ func TestHighFreqUpdates(t *testing.T) {
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)
readCountWithoutHint := readCount
// 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 = lookup.Lookup(context.Background(), now, epoch, readFunc)
value, err = algo.Lookup(context.Background(), now, epoch, readFunc)
stopwatch.Stop()
if err != nil {
t.Fatal(err)
}
@ -377,13 +480,17 @@ func TestHighFreqUpdates(t *testing.T) {
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 equal reads than %d since we provided a hint. Did %d reads.", readCountWithoutHint, readCount)
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())
for i := uint64(0); i <= 994; i++ {
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 := lookup.Lookup(context.Background(), T, lookup.NoClue, readFunc)
value, err := algo.Lookup(context.Background(), T, lookup.NoClue, readFunc)
if err != nil {
t.Fatal(err)
}
@ -395,9 +502,16 @@ func TestHighFreqUpdates(t *testing.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,23 +533,33 @@ 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)
// try to get the last value
value, err := algo.Lookup(context.Background(), now, lookup.NoClue, readFunc)
stopwatch.Stop()
if err != nil {
t.Fatal(err)
}
readCountWithoutHint := readCount
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 = lookup.Lookup(context.Background(), now, epoch, readFunc)
value, err = algo.Lookup(context.Background(), now, epoch, readFunc)
stopwatch.Stop()
if err != nil {
t.Fatal(err)
}
@ -444,10 +568,14 @@ func TestSparseUpdates(t *testing.T) {
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)
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())
})
}
}
// 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)
}

94
swarm/storage/feed/lookup/timesim_test.go Normal file
View file

@ -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
}
}