swarm/storage/feed/lookup: First LE that works

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 <nil> 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
+	}
 }

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)
+}