diff --git a/swarm/storage/feed/lookup/lookup.go b/swarm/storage/feed/lookup/lookup.go index 1642c659a8..e6f13d675e 100644 --- a/swarm/storage/feed/lookup/lookup.go +++ b/swarm/storage/feed/lookup/lookup.go @@ -20,7 +20,10 @@ so they can be found */ package lookup -import "context" +import ( + "context" + "time" +) const maxuint64 = ^uint64(0) @@ -43,7 +46,7 @@ type Algorithm func(ctx context.Context, now uint64, hint Epoch, read ReadFunc) // read() will be called on each lookup attempt // Returns an error only if read() returns an error // Returns nil if an update was not found -var Lookup Algorithm = FluzCapacitorAlgorithm +var Lookup Algorithm = LongEarthAlgorithm // ReadFunc is a handler called by Lookup each time it attempts to find a value // It should return if a value is not found @@ -180,4 +183,133 @@ func FluzCapacitorAlgorithm(ctx context.Context, now uint64, hint Epoch, read Re } 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 + } } diff --git a/swarm/storage/feed/lookup/lookup_test.go b/swarm/storage/feed/lookup/lookup_test.go index 60d77b7096..3322d2f017 100644 --- a/swarm/storage/feed/lookup/lookup_test.go +++ b/swarm/storage/feed/lookup/lookup_test.go @@ -20,7 +20,9 @@ import ( "context" "fmt" "math/rand" + "sync/atomic" "testing" + "time" "github.com/ethereum/go-ethereum/swarm/log" "github.com/ethereum/go-ethereum/swarm/storage/feed/lookup" @@ -31,6 +33,10 @@ type Data struct { Time uint64 } +func (d *Data) String() string { + return fmt.Sprintf("%d-%d", d.Payload, d.Time) +} + type Store map[lookup.EpochID]*Data func write(store Store, epoch lookup.Epoch, value *Data) { @@ -50,15 +56,21 @@ const Day = 60 * 60 * 24 const Year = Day * 365 const Month = Day * 30 -func makeReadFunc(store Store, counter *int) lookup.ReadFunc { +func makeReadFunc(store Store, counter *int32) lookup.ReadFunc { return func(ctx context.Context, epoch lookup.Epoch, now uint64) (interface{}, error) { - *counter++ + atomic.AddInt32(counter, 1) + select { + case <-time.After(1000 * time.Millisecond): + case <-ctx.Done(): + return nil, ctx.Err() + } data := store[epoch.ID()] var valueStr string if data != nil { 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 } @@ -67,9 +79,8 @@ func makeReadFunc(store Store, counter *int) lookup.ReadFunc { } func TestLookup(t *testing.T) { - store := make(Store) - readCount := 0 + var readCount int32 = 0 readFunc := makeReadFunc(store, &readCount) // write an update every month for 12 months 3 years ago and then silence for two years @@ -93,6 +104,7 @@ func TestLookup(t *testing.T) { if err != nil { t.Fatal(err) } + fmt.Printf("readcount=%d\n", readCount) readCountWithoutHint := readCount @@ -142,7 +154,7 @@ func TestLookup(t *testing.T) { func TestOneUpdateAt0(t *testing.T) { store := make(Store) - readCount := 0 + var readCount int32 = 0 readFunc := makeReadFunc(store, &readCount) now := uint64(1533903729) @@ -167,7 +179,7 @@ func TestOneUpdateAt0(t *testing.T) { func TestBadHint(t *testing.T) { store := make(Store) - readCount := 0 + var readCount int32 = 0 readFunc := makeReadFunc(store, &readCount) now := uint64(1533903729) @@ -299,7 +311,7 @@ func TestContextCancellation(t *testing.T) { func TestLookupFail(t *testing.T) { store := make(Store) - readCount := 0 + var readCount int32 = 0 readFunc := makeReadFunc(store, &readCount) now := uint64(1533903729) @@ -324,7 +336,7 @@ func TestLookupFail(t *testing.T) { func TestHighFreqUpdates(t *testing.T) { store := make(Store) - readCount := 0 + var readCount int32 = 0 readFunc := makeReadFunc(store, &readCount) now := uint64(1533903729) @@ -388,7 +400,7 @@ func TestHighFreqUpdates(t *testing.T) { func TestSparseUpdates(t *testing.T) { store := make(Store) - readCount := 0 + var readCount int32 = 0 readFunc := makeReadFunc(store, &readCount) // write an update every 5 years 3 times starting in Jan 1st 1970 and then silence @@ -513,3 +525,14 @@ 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) +}