|
| 1 | +package lookup |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "sync/atomic" |
| 6 | + "time" |
| 7 | +) |
| 8 | + |
| 9 | +type stepFunc func(ctx context.Context, t uint64, hint Epoch) interface{} |
| 10 | + |
| 11 | +// LongEarthLookaheadDelay is the headstart the lookahead gives R before it launches |
| 12 | +var LongEarthLookaheadDelay = 250 * time.Millisecond |
| 13 | + |
| 14 | +// LongEarthLookbackDelay is the headstart the lookback gives R before it launches |
| 15 | +var LongEarthLookbackDelay = 250 * time.Millisecond |
| 16 | + |
| 17 | +// LongEarthAlgorithm explores possible lookup paths in parallel, pruning paths as soon |
| 18 | +// as a more promising lookup path is found. As a result, this lookup algorithm is an order |
| 19 | +// of magnitude faster than the FluzCapacitor algorithm, but at the expense of more exploratory reads. |
| 20 | +// This algorithm works as follows. On each step, the next epoch is immediately looked up (R) |
| 21 | +// and given a head start, while two parallel "steps" are launched a short time after: |
| 22 | +// look ahead (A) is the path the algorithm would take if the R lookup returns a value, whereas |
| 23 | +// look back (B) is the path the algorithm would take if the R lookup failed. |
| 24 | +// as soon as R is actually finished, the A or B paths are pruned depending on the value of R. |
| 25 | +// if A returns earlier than R, then R and B read operations can be safely canceled, saving time. |
| 26 | +// The maximum number of active read operations is calculated as 2^(timeout/headstart). |
| 27 | +// If headstart is infinite, this algorithm behaves as FluzCapacitor. |
| 28 | +// timeout is the maximum execution time of the passed `read` function. |
| 29 | +// the two head starts can be configured by changing LongEarthLookaheadDelay or LongEarthLookbackDelay |
| 30 | +func LongEarthAlgorithm(ctx context.Context, now uint64, hint Epoch, read ReadFunc) (interface{}, error) { |
| 31 | + if hint == NoClue { |
| 32 | + hint = worstHint |
| 33 | + } |
| 34 | + |
| 35 | + var stepCounter int32 // for debugging, stepCounter allows to give an ID to each step instance |
| 36 | + |
| 37 | + errc := make(chan struct{}) // errc will help as an error shortcut signal |
| 38 | + var gerr error // in case of error, this variable will be set |
| 39 | + |
| 40 | + var step stepFunc // For efficiency, the algorithm step is defined as a closure |
| 41 | + step = func(ctxS context.Context, t uint64, last Epoch) interface{} { |
| 42 | + stepID := atomic.AddInt32(&stepCounter, 1) // give an ID to this call instance |
| 43 | + trace(stepID, "init: t=%d, last=%s", t, last.String()) |
| 44 | + var valueA, valueB, valueR interface{} |
| 45 | + |
| 46 | + // initialize the three read contexts |
| 47 | + ctxR, cancelR := context.WithCancel(ctxS) // will handle the current read operation |
| 48 | + ctxA, cancelA := context.WithCancel(ctxS) // will handle the lookahead path |
| 49 | + ctxB, cancelB := context.WithCancel(ctxS) // will handle the lookback path |
| 50 | + |
| 51 | + epoch := GetNextEpoch(last, t) // calculate the epoch to look up in this step instance |
| 52 | + |
| 53 | + // define the lookAhead function, which will follow the path as if R was successful |
| 54 | + lookAhead := func() { |
| 55 | + valueA = step(ctxA, t, epoch) // launch the next step, recursively. |
| 56 | + if valueA != nil { // if this path is successful, we don't need R or B. |
| 57 | + cancelB() |
| 58 | + cancelR() |
| 59 | + } |
| 60 | + } |
| 61 | + |
| 62 | + // define the lookBack function, which will follow the path as if R was unsuccessful |
| 63 | + lookBack := func() { |
| 64 | + if epoch.Base() == last.Base() { |
| 65 | + return |
| 66 | + } |
| 67 | + base := epoch.Base() |
| 68 | + if base == 0 { |
| 69 | + return |
| 70 | + } |
| 71 | + valueB = step(ctxB, base-1, last) |
| 72 | + } |
| 73 | + |
| 74 | + go func() { //goroutine to read the current epoch (R) |
| 75 | + defer cancelR() |
| 76 | + var err error |
| 77 | + valueR, err = read(ctxR, epoch, now) // read this epoch |
| 78 | + if valueR == nil { // if unsuccessful, cancel lookahead, otherwise cancel lookback. |
| 79 | + cancelA() |
| 80 | + } else { |
| 81 | + cancelB() |
| 82 | + } |
| 83 | + if err != nil && err != context.Canceled { |
| 84 | + gerr = err |
| 85 | + close(errc) |
| 86 | + } |
| 87 | + }() |
| 88 | + |
| 89 | + go func() { // goroutine to give a headstart to R and then launch lookahead. |
| 90 | + defer cancelA() |
| 91 | + |
| 92 | + // if we are at the lowest level or the epoch to look up equals the last one, |
| 93 | + // then we cannot lookahead (can't go lower or repeat the same lookup, this would |
| 94 | + // cause an infinite loop) |
| 95 | + if epoch.Level == LowestLevel || epoch.Equals(last) { |
| 96 | + return |
| 97 | + } |
| 98 | + |
| 99 | + // give a head start to R, or launch immediately if R finishes early enough |
| 100 | + select { |
| 101 | + case <-TimeAfter(LongEarthLookaheadDelay): |
| 102 | + lookAhead() |
| 103 | + case <-ctxR.Done(): |
| 104 | + if valueR != nil { |
| 105 | + lookAhead() // only look ahead if R was successful |
| 106 | + } |
| 107 | + case <-ctxA.Done(): |
| 108 | + } |
| 109 | + }() |
| 110 | + |
| 111 | + go func() { // goroutine to give a headstart to R and then launch lookback. |
| 112 | + defer cancelB() |
| 113 | + |
| 114 | + // give a head start to R, or launch immediately if R finishes early enough |
| 115 | + select { |
| 116 | + case <-TimeAfter(LongEarthLookbackDelay): |
| 117 | + lookBack() |
| 118 | + case <-ctxR.Done(): |
| 119 | + if valueR == nil { |
| 120 | + lookBack() // only look back in case R failed |
| 121 | + } |
| 122 | + case <-ctxB.Done(): |
| 123 | + } |
| 124 | + }() |
| 125 | + |
| 126 | + <-ctxA.Done() |
| 127 | + if valueA != nil { |
| 128 | + trace(stepID, "Returning valueA=%v", valueA) |
| 129 | + return valueA |
| 130 | + } |
| 131 | + |
| 132 | + <-ctxR.Done() |
| 133 | + if valueR != nil { |
| 134 | + trace(stepID, "Returning valueR=%v", valueR) |
| 135 | + return valueR |
| 136 | + } |
| 137 | + <-ctxB.Done() |
| 138 | + trace(stepID, "Returning valueB=%v", valueB) |
| 139 | + return valueB |
| 140 | + } |
| 141 | + |
| 142 | + var value interface{} |
| 143 | + stepCtx, cancel := context.WithCancel(ctx) |
| 144 | + |
| 145 | + go func() { // launch the root step in its own goroutine to allow cancellation |
| 146 | + defer cancel() |
| 147 | + value = step(stepCtx, now, hint) |
| 148 | + }() |
| 149 | + |
| 150 | + // wait for the algorithm to finish, but shortcut in case |
| 151 | + // of errors |
| 152 | + select { |
| 153 | + case <-stepCtx.Done(): |
| 154 | + case <-errc: |
| 155 | + cancel() |
| 156 | + return nil, gerr |
| 157 | + } |
| 158 | + |
| 159 | + if ctx.Err() != nil { |
| 160 | + return nil, ctx.Err() |
| 161 | + } |
| 162 | + |
| 163 | + if value != nil || hint == worstHint { |
| 164 | + return value, nil |
| 165 | + } |
| 166 | + |
| 167 | + // at this point the algorithm did not return a value, |
| 168 | + // so we challenge the hint given. |
| 169 | + value, err := read(ctx, hint, now) |
| 170 | + if err != nil { |
| 171 | + return nil, err |
| 172 | + } |
| 173 | + if value != nil { |
| 174 | + return value, nil // hint is valid, return it. |
| 175 | + } |
| 176 | + |
| 177 | + // hint is invalid. Invoke the algorithm |
| 178 | + // without hint. |
| 179 | + now = hint.Base() |
| 180 | + if hint.Level == HighestLevel { |
| 181 | + now-- |
| 182 | + } |
| 183 | + |
| 184 | + return LongEarthAlgorithm(ctx, now, NoClue, read) |
| 185 | +} |
0 commit comments